feat: add OTLP canary query (#4488)

* Working gRPC probe

* Fixed incorrect guessing of OTLP sample types

* Fixed lints

Author: oleg-kozlyuk-grafana

.vscode/launch.json

@@ -10,6 +10,26 @@
             "request": "launch",
             "mode": "auto",
             "program": "${workspaceFolder}/cmd/pyroscope",
+        },
+        {
+            "name": "canary-exporter",
+            "type": "go",
+            "request": "launch",
+            "mode": "auto",
+            "program": "${workspaceFolder}/cmd/profilecli",
+            "args": [
+                "canary-exporter"
+            ]
+        }
+    ],
+    "compounds": [
+        {
+            "name": "canary+pyroscope",
+            "configurations": [
+                "pyroscope",
+                "canary-exporter"
+            ],
+            "stopAll": true
         }
     ]
 }

cmd/profilecli/canary_exporter.go

@@ -160,6 +160,7 @@ func newCanaryExporter(params *canaryExporterParams) *canaryExporter {
 	}
 
 	ce.queryProbes = append(ce.queryProbes, &queryProbe{name: "query-select-merge-profile", f: ce.testSelectMergeProfile})
+	ce.queryProbes = append(ce.queryProbes, &queryProbe{name: "query-select-merge-otlp-profile", f: ce.testSelectMergeOTLPProfile})
 
 	if params.QueryProbeSet == "all" {
 		ce.queryProbes = append(ce.queryProbes, &queryProbe{"query-profile-types", ce.testProfileTypes})
@@ -256,7 +257,10 @@ func (ce *canaryExporter) doTrace(ctx context.Context, probeName string) (rCtx c
 
 	return rCtx, func(result bool) {
 		// At this point body is fully read and we can write end time.
-		tt.current.end = time.Now()
+		// Note: tt.current may be nil for non-HTTP probes (e.g., gRPC)
+		if tt.current != nil {
+			tt.current.end = time.Now()
+		}
 
 		// record body size
 		ce.metrics.bodyUncompressedLength.WithLabelValues(probeName).Set(float64(tt.bodySize.Load()))
@@ -310,11 +314,46 @@ func (ce *canaryExporter) doTrace(ctx context.Context, probeName string) (rCtx c
 func (ce *canaryExporter) testPyroscopeCell(ctx context.Context) error {
 	now := time.Now()
 
-	// ingest a fake profile
+	// Run all ingest probes
+	var ingestErrors multierror.MultiError
+
+	// ingest a fake profile using the original method
 	if err := ce.runProbe(ctx, "ingest", func(rCtx context.Context) error {
 		return ce.testIngestProfile(rCtx, now)
 	}); err != nil {
-		return fmt.Errorf("error during ingestion: %w", err)
+		ingestErrors.Add(fmt.Errorf("error during standard ingestion: %w", err))
+	}
+
+	// ingest via OTLP gRPC
+	if err := ce.runProbe(ctx, "ingest-otlp-grpc", func(rCtx context.Context) error {
+		return ce.testIngestOTLPGrpc(rCtx, now)
+	}); err != nil {
+		ingestErrors.Add(fmt.Errorf("error during OTLP gRPC ingestion: %w", err))
+	}
+
+	// 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))
+	// }
+
+	// 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))
+	// }
+
+	// Report ingestion errors if any
+	if ingestErrors.Err() != nil {
+		for _, line := range strings.Split(ingestErrors.Err().Error(), "\n") {
+			level.Error(logger).Log("msg", "ingestion error", "err", line)
+		}
 	}
 
 	if ce.params.TestDelay > 0 {

cmd/profilecli/canary_exporter_probes.go

@@ -19,6 +19,12 @@ import (
 	gprofile "github.com/google/pprof/profile"
 	"github.com/google/uuid"
 	"github.com/pkg/errors"
+	profilesv1 "go.opentelemetry.io/proto/otlp/collector/profiles/v1development"
+	commonv1 "go.opentelemetry.io/proto/otlp/common/v1"
+	otlpprofiles "go.opentelemetry.io/proto/otlp/profiles/v1development"
+	resourcev1 "go.opentelemetry.io/proto/otlp/resource/v1"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 
 	googlev1 "github.com/grafana/pyroscope/api/gen/proto/go/google/v1"
 	pushv1 "github.com/grafana/pyroscope/api/gen/proto/go/push/v1"
@@ -31,6 +37,7 @@ import (
 
 const (
 	profileTypeID             = "deadmans_switch:made_up:profilos:made_up:profilos"
+	otlpProfileTypeID         = "otlp_test:otlp_test:count:otlp_test:samples"
 	canaryExporterServiceName = "pyroscope-canary-exporter"
 )
 
@@ -74,6 +81,140 @@ func (ce *canaryExporter) testIngestProfile(ctx context.Context, now time.Time)
 	return nil
 }
 
+// generateOTLPProfile creates an OTLP profile with the specified ingestion method label
+func (ce *canaryExporter) generateOTLPProfile(now time.Time, ingestionMethod string) *profilesv1.ExportProfilesServiceRequest {
+	// Sanitize the ingestion method label value by replacing "/" with "_"
+	sanitizedMethod := strings.ReplaceAll(ingestionMethod, "/", "_")
+
+	// Create the profile dictionary with custom profile type similar to pprof probe
+	dictionary := &otlpprofiles.ProfilesDictionary{
+		StringTable: []string{
+			"",                 // 0: empty string
+			"otlp_test",        // 1
+			"samples",          // 2
+			"count",            // 3
+			"func1",            // 4
+			"func2",            // 5
+			"ingestion_method", // 6
+			sanitizedMethod,    // 7
+		},
+		MappingTable: []*otlpprofiles.Mapping{
+			{}, // 0: empty mapping (required null entry)
+		},
+		FunctionTable: []*otlpprofiles.Function{
+			{NameStrindex: 0}, // 0: empty
+			{NameStrindex: 4}, // 1: func1
+			{NameStrindex: 5}, // 2: func2
+		},
+		LocationTable: []*otlpprofiles.Location{
+			{Line: []*otlpprofiles.Line{{FunctionIndex: 1}}}, // 0: func1
+			{Line: []*otlpprofiles.Line{{FunctionIndex: 2}}}, // 1: func2
+		},
+		StackTable: []*otlpprofiles.Stack{
+			{LocationIndices: []int32{}},     // 0: empty (required null entry)
+			{LocationIndices: []int32{1, 0}}, // 1: func2, func1 stack
+			{LocationIndices: []int32{0}},    // 2: func1 stack
+		},
+	}
+
+	// Create profile with two samples matching the original pprof profile
+	profile := &otlpprofiles.Profile{
+		TimeUnixNano: uint64(now.UnixNano()),
+		DurationNano: 0,
+		Period:       1,
+		SampleType: &otlpprofiles.ValueType{
+			TypeStrindex: 1, // "otlp_test"
+			UnitStrindex: 3, // "count"
+		},
+		PeriodType: &otlpprofiles.ValueType{
+			TypeStrindex: 1, // "otlp_test"
+			UnitStrindex: 2, // "samples"
+		},
+		Sample: []*otlpprofiles.Sample{
+			{
+				// func1>func2 with value 10
+				StackIndex: 1, // stack_table[1]
+				Values:     []int64{10},
+			},
+			{
+				// func1 with value 20
+				StackIndex: 2, // stack_table[2]
+				Values:     []int64{20},
+			},
+		},
+	}
+
+	// Create the resource attributes
+	resourceAttrs := []*commonv1.KeyValue{
+		{
+			Key:   "service.name",
+			Value: &commonv1.AnyValue{Value: &commonv1.AnyValue_StringValue{StringValue: canaryExporterServiceName}},
+		},
+		{
+			Key:   "job",
+			Value: &commonv1.AnyValue{Value: &commonv1.AnyValue_StringValue{StringValue: "canary-exporter"}},
+		},
+		{
+			Key:   "instance",
+			Value: &commonv1.AnyValue{Value: &commonv1.AnyValue_StringValue{StringValue: ce.hostname}},
+		},
+		{
+			Key:   "ingestion_method",
+			Value: &commonv1.AnyValue{Value: &commonv1.AnyValue_StringValue{StringValue: sanitizedMethod}},
+		},
+	}
+
+	// Create the OTLP request
+	req := &profilesv1.ExportProfilesServiceRequest{
+		Dictionary: dictionary,
+		ResourceProfiles: []*otlpprofiles.ResourceProfiles{
+			{
+				Resource: &resourcev1.Resource{
+					Attributes: resourceAttrs,
+				},
+				ScopeProfiles: []*otlpprofiles.ScopeProfiles{
+					{
+						Scope: &commonv1.InstrumentationScope{
+							Name: "pyroscope-canary-exporter",
+						},
+						Profiles: []*otlpprofiles.Profile{profile},
+					},
+				},
+			},
+		},
+	}
+
+	return req
+}
+
+func (ce *canaryExporter) testIngestOTLPGrpc(ctx context.Context, now time.Time) error {
+	// Generate the OTLP profile with the appropriate ingestion method label
+	req := ce.generateOTLPProfile(now, "otlp/grpc")
+
+	// Create gRPC connection
+	grpcAddr := strings.TrimPrefix(ce.params.URL, "http://")
+	grpcAddr = strings.TrimPrefix(grpcAddr, "https://")
+
+	conn, err := grpc.NewClient(grpcAddr,
+		grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		return fmt.Errorf("failed to connect to gRPC server: %w", err)
+	}
+	defer conn.Close()
+
+	// Create OTLP profiles service client
+	client := profilesv1.NewProfilesServiceClient(conn)
+
+	// Send the profile
+	_, err = client.Export(ctx, req)
+	if err != nil {
+		return fmt.Errorf("failed to export OTLP profile via gRPC: %w", err)
+	}
+
+	level.Info(logger).Log("msg", "successfully ingested OTLP profile via gRPC")
+	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(),
@@ -128,6 +269,65 @@ func (ce *canaryExporter) testSelectMergeProfile(ctx context.Context, now time.T
 	return nil
 }
 
+func (ce *canaryExporter) testSelectMergeOTLPProfile(ctx context.Context, now time.Time) error {
+	// Query specifically for OTLP gRPC ingested profiles using the custom profile type
+	//labelSelector := fmt.Sprintf(`{service_name="%s", job="canary-exporter", instance="%s"}`, canaryExporterServiceName, ce.hostname)
+
+	respQuery, err := ce.params.queryClient().SelectMergeProfile(ctx, connect.NewRequest(&querierv1.SelectMergeProfileRequest{
+		Start:         now.UnixMilli(),
+		End:           now.Add(5 * time.Second).UnixMilli(),
+		LabelSelector: ce.createLabelSelector(),
+		ProfileTypeID: otlpProfileTypeID,
+	}))
+	if err != nil {
+		return fmt.Errorf("failed to query OTLP profile: %w", err)
+	}
+
+	buf, err := respQuery.Msg.MarshalVT()
+	if err != nil {
+		return errors.Wrap(err, "failed to marshal protobuf")
+	}
+
+	gp, err := gprofile.Parse(bytes.NewReader(buf))
+	if err != nil {
+		return errors.Wrap(err, "failed to parse profile")
+	}
+
+	// Verify the expected stacktraces from the OTLP profile
+	expected := map[string]int64{
+		"func2>func1": 10,
+		"func1":       20,
+	}
+	actual := make(map[string]int64)
+
+	var sb strings.Builder
+	for _, s := range gp.Sample {
+		sb.Reset()
+		for _, loc := range s.Location {
+			if sb.Len() != 0 {
+				_, err := sb.WriteRune('>')
+				if err != nil {
+					return err
+				}
+			}
+			for _, line := range loc.Line {
+				_, err := sb.WriteString(line.Function.Name)
+				if err != nil {
+					return err
+				}
+			}
+		}
+		actual[sb.String()] = actual[sb.String()] + s.Value[0]
+	}
+
+	if diff := cmp.Diff(expected, actual); diff != "" {
+		return fmt.Errorf("OTLP profile query mismatch (-expected, +actual):\n%s", diff)
+	}
+
+	level.Info(logger).Log("msg", "successfully queried OTLP profile via gRPC")
+	return nil
+}
+
 func (ce *canaryExporter) testProfileTypes(ctx context.Context, now time.Time) error {
 	respQuery, err := ce.params.queryClient().ProfileTypes(ctx, connect.NewRequest(&querierv1.ProfileTypesRequest{
 		Start: now.UnixMilli(),

pkg/ingester/otlp/convert.go

@@ -3,6 +3,7 @@ package otlp
 import (
 	"encoding/hex"
 	"fmt"
+	"strings"
 	"time"
 
 	otelProfile "go.opentelemetry.io/proto/otlp/profiles/v1development"
@@ -144,14 +145,23 @@ func newProfileBuilder(src *otelProfile.Profile, dictionary *otelProfile.Profile
 				}
 			}
 			res.sampleProcessingTypes[i] = sampleConversionTypeSamplesToNanos
-		}
-		// Identify off cpu profiles
-		if profileType == "events:nanoseconds::" && len(res.dst.SampleType) == 1 {
+		} else if profileType == "events:nanoseconds::" && len(res.dst.SampleType) == 1 { // Identify off-CPU profiles
+
 			res.sampleProcessingTypes[i] = sampleConversionTypeSumEvents
 			res.name = &typesv1.LabelPair{
 				Name:  pyromodel.LabelNameProfileName,
 				Value: pyromodel.ProfileNameOffCpu,
 			}
+		} else { // Custom profile type
+			// Try to extract profile name from the type, e.g. "wall:time:cpu:milliseconds" -> "wall"
+			parts := strings.Split(profileType, `:`)
+			if len(parts) >= 3 {
+				res.name = &typesv1.LabelPair{
+					Name:  pyromodel.LabelNameProfileName,
+					Value: parts[2],
+				}
+				res.sampleProcessingTypes[i] = sampleConversionTypeNone
+			}
 		}
 	}
 	if res.name == nil {