Add OTLP HTTP/json and HTTP/protobuf ingestion (#4495)

* Added HTTP/json and HTTP/protobuf OTLP ingestion

* Add http/json+http/protobuf ingestion with canary

* Updated endpoint path to match OTLP

Author: oleg-kozlyuk-grafana

cmd/profilecli/canary_exporter.go

@@ -333,21 +333,19 @@ func (ce *canaryExporter) testPyroscopeCell(ctx context.Context) error {
 
 	// ingest via OTLP HTTP/JSON
 	// Note: HTTP endpoints are not yet implemented in Pyroscope (see pkg/api/api.go:204-205)
-	// Uncomment when /v1/profiles endpoint is available
-	// if err := ce.runProbe(ctx, "ingest-otlp-http-json", func(rCtx context.Context) error {
-	// 	return ce.testIngestOTLPHttpJson(rCtx, now)
-	// }); err != nil {
-	// 	ingestErrors.Add(fmt.Errorf("error during OTLP HTTP/JSON ingestion: %w", err))
-	// }
+	if err := ce.runProbe(ctx, "ingest-otlp-http-json", func(rCtx context.Context) error {
+		return ce.testIngestOTLPHttpJson(rCtx, now)
+	}); err != nil {
+		ingestErrors.Add(fmt.Errorf("error during OTLP HTTP/JSON ingestion: %w", err))
+	}
 
 	// ingest via OTLP HTTP/Protobuf
 	// Note: HTTP endpoints are not yet implemented in Pyroscope (see pkg/api/api.go:204-205)
-	// Uncomment when /v1/profiles endpoint is available
-	// if err := ce.runProbe(ctx, "ingest-otlp-http-protobuf", func(rCtx context.Context) error {
-	// 	return ce.testIngestOTLPHttpProtobuf(rCtx, now)
-	// }); err != nil {
-	// 	ingestErrors.Add(fmt.Errorf("error during OTLP HTTP/Protobuf ingestion: %w", err))
-	// }
+	if err := ce.runProbe(ctx, "ingest-otlp-http-protobuf", func(rCtx context.Context) error {
+		return ce.testIngestOTLPHttpProtobuf(rCtx, now)
+	}); err != nil {
+		ingestErrors.Add(fmt.Errorf("error during OTLP HTTP/Protobuf ingestion: %w", err))
+	}
 
 	// Report ingestion errors if any
 	if ingestErrors.Err() != nil {

cmd/profilecli/canary_exporter_probes.go

@@ -25,6 +25,8 @@ import (
 	resourcev1 "go.opentelemetry.io/proto/otlp/resource/v1"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
 
 	googlev1 "github.com/grafana/pyroscope/api/gen/proto/go/google/v1"
 	pushv1 "github.com/grafana/pyroscope/api/gen/proto/go/push/v1"
@@ -215,6 +217,86 @@ func (ce *canaryExporter) testIngestOTLPGrpc(ctx context.Context, now time.Time)
 	return nil
 }
 
+func (ce *canaryExporter) testIngestOTLPHttpJson(ctx context.Context, now time.Time) error {
+	// Generate the OTLP profile with the appropriate ingestion method label
+	req := ce.generateOTLPProfile(now, "otlp/http/json")
+
+	// Marshal to JSON
+	jsonData, err := protojson.Marshal(req)
+	if err != nil {
+		return fmt.Errorf("failed to marshal OTLP profile to JSON: %w", err)
+	}
+
+	// Create HTTP request using the instrumented client
+	url := ce.params.URL + "/v1development/profiles"
+	httpReq, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(jsonData))
+	if err != nil {
+		return fmt.Errorf("failed to create HTTP request: %w", err)
+	}
+	httpReq.Header.Set("Content-Type", "application/json")
+
+	// Send the request using the instrumented client (ce.params.client is set by doTrace)
+	resp, err := ce.params.client.Do(httpReq)
+	if err != nil {
+		return fmt.Errorf("failed to send HTTP request: %w", err)
+	}
+	defer resp.Body.Close()
+
+	// Read the body to ensure the transport is fully traced
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return fmt.Errorf("failed to read response body: %w", err)
+	}
+
+	// Check response status
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("unexpected status code %d: %s", resp.StatusCode, string(body))
+	}
+
+	level.Info(logger).Log("msg", "successfully ingested OTLP profile via HTTP/JSON")
+	return nil
+}
+
+func (ce *canaryExporter) testIngestOTLPHttpProtobuf(ctx context.Context, now time.Time) error {
+	// Generate the OTLP profile with the appropriate ingestion method label
+	req := ce.generateOTLPProfile(now, "otlp/http/protobuf")
+
+	// Marshal to protobuf
+	protoData, err := proto.Marshal(req)
+	if err != nil {
+		return fmt.Errorf("failed to marshal OTLP profile to protobuf: %w", err)
+	}
+
+	// Create HTTP request using the instrumented client
+	url := ce.params.URL + "/v1development/profiles"
+	httpReq, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(protoData))
+	if err != nil {
+		return fmt.Errorf("failed to create HTTP request: %w", err)
+	}
+	httpReq.Header.Set("Content-Type", "application/x-protobuf")
+
+	// Send the request using the instrumented client (ce.params.client is set by doTrace)
+	resp, err := ce.params.client.Do(httpReq)
+	if err != nil {
+		return fmt.Errorf("failed to send HTTP request: %w", err)
+	}
+	defer resp.Body.Close()
+
+	// Read the body to ensure the transport is fully traced
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return fmt.Errorf("failed to read response body: %w", err)
+	}
+
+	// Check response status
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("unexpected status code %d: %s", resp.StatusCode, string(body))
+	}
+
+	level.Info(logger).Log("msg", "successfully ingested OTLP profile via HTTP/Protobuf")
+	return nil
+}
+
 func (ce *canaryExporter) testSelectMergeProfile(ctx context.Context, now time.Time) error {
 	respQuery, err := ce.params.queryClient().SelectMergeProfile(ctx, connect.NewRequest(&querierv1.SelectMergeProfileRequest{
 		Start:         now.UnixMilli(),
@@ -295,8 +377,8 @@ func (ce *canaryExporter) testSelectMergeOTLPProfile(ctx context.Context, now ti
 
 	// Verify the expected stacktraces from the OTLP profile
 	expected := map[string]int64{
-		"func2>func1": 10,
-		"func1":       20,
+		"func2>func1": 30, // 10 samples from each of the 3 ingestion methods
+		"func1":       60, // 20 samples * 3
 	}
 	actual := make(map[string]int64)
 

pkg/api/api.go

@@ -201,8 +201,7 @@ func (a *API) RegisterDistributor(d *distributor.Distributor, limits *validation
 	})
 
 	a.RegisterRoute("/opentelemetry.proto.collector.profiles.v1development.ProfilesService/Export", otlpHandler, writePathOpts...)
-	// TODO(@petethepig): implement http/protobuf and http/json support
-	// a.RegisterRoute("/v1/profiles", otlpHandler, true, true, "POST")
+	a.RegisterRoute("/v1development/profiles", otlpHandler, writePathOpts...)
 }
 
 // RegisterMemberlistKV registers the endpoints associated with the memberlist KV store.

pkg/ingester/otlp/ingest_handler.go

@@ -3,12 +3,14 @@ package otlp
 import (
 	"context"
 	"fmt"
+	"io"
 	"net/http"
 	"strings"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/keepalive"
+	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/go-kit/log"
@@ -61,10 +63,14 @@ func NewOTLPIngestHandler(cfg server.Config, svc PushService, l log.Logger, me b
 			return
 		}
 
-		// Handle HTTP requests (if we want to support HTTP/Protobuf in future)
-		//if r.URL.Path == "/v1/profiles" {}
+		// Handle HTTP/Protobuf and HTTP/Binary requests
+		contentType := r.Header.Get("Content-Type")
+		if contentType == "application/json" || contentType == "application/x-protobuf" || contentType == "application/protobuf" {
+			h.handleHTTPProtobuf(w, r)
+			return
+		}
 
-		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+		http.Error(w, fmt.Sprintf("Unsupported Content-Type: %s", contentType), http.StatusUnsupportedMediaType)
 	})
 
 	return h
@@ -102,6 +108,76 @@ func (h *ingestHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	h.handler.ServeHTTP(w, r)
 }
 
+func (h *ingestHandler) handleHTTPProtobuf(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+
+	// Read the request body - we need to read it all for protobuf unmarshaling
+	// Note: Protobuf wire format requires reading the entire message to determine field boundaries
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		level.Error(h.log).Log("msg", "failed to read request body", "err", err)
+		http.Error(w, "Failed to read request body", http.StatusBadRequest)
+		return
+	}
+
+	// Unmarshal the protobuf request
+	req := &pprofileotlp.ExportProfilesServiceRequest{}
+
+	if r.Header.Get("Content-Type") == "application/json" {
+		if err := protojson.Unmarshal(body, req); err != nil {
+			level.Error(h.log).Log("msg", "failed to unmarshal JSON request", "err", err)
+			http.Error(w, "Failed to parse JSON request", http.StatusBadRequest)
+			return
+		}
+	} else {
+		if err := proto.Unmarshal(body, req); err != nil {
+			level.Error(h.log).Log("msg", "failed to unmarshal protobuf request", "err", err)
+			http.Error(w, "Failed to parse protobuf request", http.StatusBadRequest)
+			return
+		}
+	}
+
+	ctx := r.Context()
+	// Process the request using the existing Export method
+	// Injects multitenancy info into context if needed
+	resp, err := h.Export(ctx, req)
+	if err != nil {
+		level.Error(h.log).Log("msg", "failed to process profiles", "err", err)
+		// Convert gRPC status to HTTP status
+		st, ok := status.FromError(err)
+		if ok {
+			switch st.Code() {
+			case codes.InvalidArgument:
+				http.Error(w, st.Message(), http.StatusBadRequest)
+			case codes.Unauthenticated:
+				http.Error(w, st.Message(), http.StatusUnauthorized)
+			case codes.PermissionDenied:
+				http.Error(w, st.Message(), http.StatusForbidden)
+			default:
+				http.Error(w, st.Message(), http.StatusInternalServerError)
+			}
+		} else {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+		}
+		return
+	}
+
+	// Marshal the response
+	respBytes, err := proto.Marshal(resp)
+	if err != nil {
+		level.Error(h.log).Log("msg", "failed to marshal response", "err", err)
+		http.Error(w, "Failed to marshal response", http.StatusInternalServerError)
+		return
+	}
+
+	// Write the response
+	w.Header().Set("Content-Type", "application/x-protobuf")
+	w.WriteHeader(http.StatusOK)
+	if _, err := w.Write(respBytes); err != nil {
+		level.Error(h.log).Log("msg", "failed to write response", "err", err)
+	}
+}
+
 func (h *ingestHandler) Export(ctx context.Context, er *pprofileotlp.ExportProfilesServiceRequest) (*pprofileotlp.ExportProfilesServiceResponse, error) {
 	// TODO: @petethepig This logic is copied from util.AuthenticateUser and should be refactored into a common function
 	// Extracts user ID from the request metadata and returns and injects the user ID in the context