Extend profile schema to support spans (#2508)

Author: kolesnikovae

cmd/pyroscope/help-all.txt.tmpl

@@ -881,6 +881,8 @@ Usage of ./pyroscope:
     	Comma-separated list of Pyroscope modules to load. The alias 'all' can be used in the list to load a number of core modules and will enable single-binary mode.  (default all)
   -tracing.enabled
     	Set to false to disable tracing. (default true)
+  -tracing.profiling-enabled
+    	[experimental] Set to true to enable profiling integration.
   -usage-stats.enabled
     	Enable anonymous usage reporting. (default true)
   -validation.max-label-names-per-series int

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

@@ -110,6 +110,10 @@ limits:
   # CLI flag: -validation.max-label-names-per-series
   [max_label_names_per_series: <int> | default = 30]
 
+  # Maximum number of sessions per series. 0 to disable.
+  # CLI flag: -validation.max-sessions-per-series
+  [max_sessions_per_series: <int> | default = 0]
+
   # Maximum size of a profile in bytes. This is based off the uncompressed size.
   # 0 to disable.
   # CLI flag: -validation.max-profile-size-bytes
@@ -714,7 +718,8 @@ grpc_tls_config:
 # CLI flag: -server.grpc-max-send-msg-size-bytes
 [grpc_server_max_send_msg_size: <int> | default = 4194304]
 
-# Limit on the number of concurrent streams for gRPC calls (0 = unlimited)
+# Limit on the number of concurrent streams for gRPC calls per client connection
+# (0 = unlimited)
 # CLI flag: -server.grpc-max-concurrent-streams
 [grpc_server_max_concurrent_streams: <int> | default = 100]
 

examples/golang-push/rideshare/go.mod

@@ -5,7 +5,7 @@ go 1.17
 require (
 	github.com/agoda-com/opentelemetry-logs-go v0.4.1-0.20231002083708-3defe69e69e4
 	github.com/grafana/pyroscope-go v1.0.4
-	github.com/pyroscope-io/otel-profiling-go v0.4.1-0.20220827003940-ddd28219c77e
+	github.com/grafana/pyroscope-go/otelpyroscope v0.0.0-20231017042744-3f66066f8d6c
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.42.0
 	go.opentelemetry.io/otel v1.19.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.19.0

examples/golang-push/rideshare/go.sum

@@ -844,7 +844,6 @@ github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2
 github.com/go-latex/latex v0.0.0-20210118124228-b3d85cf34e07/go.mod h1:CO1AlKB2CSIqUrmQPqA0gdRIlnLEY0gK5JGjh37zN5U=
 github.com/go-latex/latex v0.0.0-20210823091927-c0d11ff05a81/go.mod h1:SX0U8uGpxhq9o2S/CELCSUxEWWAuoCUcVCQWv7G2OCk=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
-github.com/go-logr/logr v1.2.3/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
 github.com/go-logr/logr v1.2.4 h1:g01GSCwiDw2xSZfjJ2/T9M+S6pFdcNtFYsp+Y43HYDQ=
 github.com/go-logr/logr v1.2.4/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
@@ -962,6 +961,8 @@ github.com/grafana/pyroscope-go v1.0.4 h1:oyQX0BOkL+iARXzHuCdIF5TQ7/sRSel1YFViMH
 github.com/grafana/pyroscope-go v1.0.4/go.mod h1:0d7ftwSMBV/Awm7CCiYmHQEG8Y44Ma3YSjt+nWcWztY=
 github.com/grafana/pyroscope-go/godeltaprof v0.1.4 h1:mDsJ3ngul7UfrHibGQpV66PbZ3q1T8glz/tK3bQKKEk=
 github.com/grafana/pyroscope-go/godeltaprof v0.1.4/go.mod h1:1HSPtjU8vLG0jE9JrTdzjgFqdJ/VgN7fvxBNq3luJko=
+github.com/grafana/pyroscope-go/otelpyroscope v0.0.0-20231017042744-3f66066f8d6c h1:5FS9csedLnPXy11nCpr7LfIyhok+cVxUYNqSTDq1bzM=
+github.com/grafana/pyroscope-go/otelpyroscope v0.0.0-20231017042744-3f66066f8d6c/go.mod h1:OhQkRYO2Iepoi2sRGUhyMp0tq1w57idp1/sz6OZGAA4=
 github.com/grpc-ecosystem/grpc-gateway v1.16.0 h1:gmcG1KaJ57LophUzW0Hy8NmPhnMZb4M0+kPpLofRdBo=
 github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.7.0/go.mod h1:hgWBS7lorOAVIJEQMi4ZsPv9hVvWI6+ch50m39Pf2Ks=
@@ -1017,8 +1018,6 @@ github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:
 github.com/prometheus/client_model v0.2.0/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/client_model v0.3.0/go.mod h1:LDGWKZIo7rky3hgvBe+caln+Dr3dPggB5dvjtD7w9+w=
 github.com/prometheus/client_model v0.4.0/go.mod h1:oMQmHW1/JoDwqLtg57MGgP/Fb1CJEYF2imWWhWtMkYU=
-github.com/pyroscope-io/otel-profiling-go v0.4.1-0.20220827003940-ddd28219c77e h1:gUdiXKQ2a6NehosBbAIZUHxBjEvzRDxjV4xn0QKcSrs=
-github.com/pyroscope-io/otel-profiling-go v0.4.1-0.20220827003940-ddd28219c77e/go.mod h1:y3pdNLRjeZi34yk7Fc1ema4SWtBEpE7Zd/rMdSfQTa8=
 github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
@@ -1067,7 +1066,6 @@ go.opencensus.io v0.23.0/go.mod h1:XItmlyltB5F7CS4xOC1DcqMoFqwtC6OG2xF7mCv7P7E=
 go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.42.0 h1:pginetY7+onl4qN1vl0xW/V/v6OBZ0vVdH+esuJgvmM=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.42.0/go.mod h1:XiYsayHc36K3EByOO6nbAXnAWbrUxdjUROCEeeROOH8=
-go.opentelemetry.io/otel v1.9.0/go.mod h1:np4EoPGzoPs3O67xUVNoPPcmSvsfOxNlNA4F4AC+0Eo=
 go.opentelemetry.io/otel v1.16.0/go.mod h1:vl0h9NUa1D5s1nv3A5vZOYWn8av4K8Ml6JDeHrT/bx4=
 go.opentelemetry.io/otel v1.19.0 h1:MuS/TNf4/j4IXsZuJegVzI1cwut7Qc00344rgH7p8bs=
 go.opentelemetry.io/otel v1.19.0/go.mod h1:i0QyjOq3UPoTzff0PJB2N66fb4S0+rSbSB15/oyH9fY=
@@ -1082,7 +1080,6 @@ go.opentelemetry.io/otel/metric v1.19.0 h1:aTzpGtV0ar9wlV4Sna9sdJyII5jTVJEvKETPi
 go.opentelemetry.io/otel/metric v1.19.0/go.mod h1:L5rUsV9kM1IxCj1MmSdS+JQAcVm319EUrDVLrt7jqt8=
 go.opentelemetry.io/otel/sdk v1.19.0 h1:6USY6zH+L8uMH8L3t1enZPR3WFEmSTADlqldyHtJi3o=
 go.opentelemetry.io/otel/sdk v1.19.0/go.mod h1:NedEbbS4w3C6zElbLdPJKOpJQOrGUJ+GfzpjUvI0v1A=
-go.opentelemetry.io/otel/trace v1.9.0/go.mod h1:2737Q0MuG8q1uILYm2YYVkAyLtOofiTNGg6VODnOiPo=
 go.opentelemetry.io/otel/trace v1.16.0/go.mod h1:Yt9vYq1SdNz3xdjZZK7wcXv1qv2pwLkqr2QVwea0ef0=
 go.opentelemetry.io/otel/trace v1.19.0 h1:DFVQmlVbfVeOuBRrwdtaehRrWiL1JoVs9CPIQ1Dzxpg=
 go.opentelemetry.io/otel/trace v1.19.0/go.mod h1:mfaSyvGyEJEI0nyV2I4qhNQnbBOUUmYZpYojqMnX2vo=

examples/golang-push/rideshare/loadgen.go

@@ -10,12 +10,13 @@ import (
 	"strconv"
 	"time"
 
-	"rideshare/rideshare"
-
+	"github.com/grafana/pyroscope-go/otelpyroscope"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/propagation"
+
+	"rideshare/rideshare"
 )
 
 var hosts = []string{}
@@ -67,7 +68,7 @@ func main() {
 	// Set the Tracer Provider and the W3C Trace Context propagator as globals.
 	// We wrap the tracer provider to also annotate goroutines with Span ID so
 	// that pprof would add corresponding labels to profiling samples.
-	otel.SetTracerProvider(tp)
+	otel.SetTracerProvider(otelpyroscope.NewTracerProvider(tp))
 
 	// Register the trace c ontext and baggage propagators so data is propagated across services/processes.
 	otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(

examples/golang-push/rideshare/main.go

@@ -14,7 +14,7 @@ import (
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 
 	otellogs "github.com/agoda-com/opentelemetry-logs-go"
-	otelpyroscope "github.com/pyroscope-io/otel-profiling-go"
+	"github.com/grafana/pyroscope-go/otelpyroscope"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/propagation"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
@@ -92,12 +92,7 @@ func setupOTEL(c rideshare.Config) (tp *sdktrace.TracerProvider, err error) {
 	// Set the Tracer Provider and the W3C Trace Context propagator as globals.
 	// We wrap the tracer provider to also annotate goroutines with Span ID so
 	// that pprof would add corresponding labels to profiling samples.
-	otel.SetTracerProvider(otelpyroscope.NewTracerProvider(tp,
-		otelpyroscope.WithPyroscopeURL(c.PyroscopeServerAddress),
-		otelpyroscope.WithAppName(c.AppName),
-		otelpyroscope.WithRootSpanOnly(true),
-		otelpyroscope.WithAddSpanName(true),
-	))
+	otel.SetTracerProvider(otelpyroscope.NewTracerProvider(tp))
 
 	// Register the trace context and baggage propagators so data is propagated across services/processes.
 	otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(

examples/tracing/tempo/README.md

@@ -30,12 +30,8 @@ Pyroscope and the demo application will be built from the current branch.  After
 4. Specify proper `start` and `end` timestamps (milliseconds).
 
 Please note that the pyroscope SDK is configured to only record root spans, therefore stack trace samples
-of child spans are included into the root span profile. This is controlled by [`WithRootSpanOnly`](../../golang-push/rideshare/main.go#L98) option.
-
-In the example we use our old SDK which also adds the `pyroscope.profile.id` attribute – although it
-might be convenient in some cases, we have no intent to maintain it. At the moment there is no reliable
-way to determine if any profiling data has been collected for a specific span, therefore this
-label may mislead.
+of child spans are included into the root span profile. These spans are marked with `pyroscope.profiling.enabled`
+attribute set to `true`.
 
 #### cURL
 

examples/tracing/tempo/docker-compose.yml

@@ -51,9 +51,9 @@ services:
 
   tempo:
     image: grafana/tempo:latest
-    command: [ "-config.file=/etc/tempo.yaml" ]
+    command: [ "-config.file=/etc/tempo.yml" ]
     volumes:
-      - ./tempo/tempo.yaml:/etc/tempo.yaml
+      - ./tempo/tempo.yml:/etc/tempo.yml
     ports:
       - "14268:14268"  # jaeger ingest
       - "3200:3200"   # tempo
@@ -65,6 +65,13 @@ services:
   pyroscope:
     ports:
       - '4040:4040'
+    environment:
+      JAEGER_AGENT_HOST: tempo
+      JAEGER_SAMPLER_TYPE: const
+      JAEGER_SAMPLER_PARAM: 1
+    volumes:
+      - ./pyroscope/pyroscope.yml:/etc/pyroscope.yml
+    command: [ "-config.file=/etc/pyroscope.yml" ]
     build:
       context: '../../../'
       dockerfile: cmd/pyroscope/Dockerfile

examples/tracing/tempo/pyroscope/pyroscope.yml

@@ -0,0 +1,6 @@
+tracing:
+  enabled: true
+  profiling_enabled: true
+
+pyroscopedb:
+  max_block_duration: 5m

examples/tracing/tempo/tempo/tempo.yaml renamed to examples/tracing/tempo/tempo/tempo.yml

@@ -30,6 +30,7 @@ import (
 	wwtracing "github.com/grafana/dskit/tracing"
 	"github.com/grafana/pyroscope-go"
 	grpcgw "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
+	"github.com/opentracing/opentracing-go"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/version"
 	"github.com/samber/lo"
@@ -53,6 +54,7 @@ import (
 	"github.com/grafana/pyroscope/pkg/usagestats"
 	"github.com/grafana/pyroscope/pkg/util"
 	"github.com/grafana/pyroscope/pkg/util/cli"
+	"github.com/grafana/pyroscope/pkg/util/spanprofiler"
 	"github.com/grafana/pyroscope/pkg/validation"
 	"github.com/grafana/pyroscope/pkg/validation/exporter"
 )
@@ -250,6 +252,9 @@ func New(cfg Config) (*Phlare, error) {
 		if err != nil {
 			level.Error(logger).Log("msg", "error in initializing tracing. tracing will not be enabled", "err", err)
 		}
+		if cfg.Tracing.ProfilingEnabled {
+			opentracing.SetGlobalTracer(spanprofiler.NewTracer(opentracing.GlobalTracer()))
+		}
 		phlare.tracer = trace
 	}
 

pkg/phlare/phlare.go

@@ -16,6 +16,7 @@ import (
 	ingestv1 "github.com/grafana/pyroscope/api/gen/proto/go/ingester/v1"
 	typesv1 "github.com/grafana/pyroscope/api/gen/proto/go/types/v1"
 	"github.com/grafana/pyroscope/pkg/iter"
+	"github.com/grafana/pyroscope/pkg/model"
 	"github.com/grafana/pyroscope/pkg/objstore/providers/filesystem"
 	"github.com/grafana/pyroscope/pkg/phlaredb/block"
 	"github.com/grafana/pyroscope/pkg/phlaredb/tsdb/index"
@@ -147,6 +148,59 @@ func TestBlockCompatability(t *testing.T) {
 	}
 }
 
+func TestBlockCompatability_SelectMergeSpans(t *testing.T) {
+	path := "./block/testdata/"
+	bucket, err := filesystem.NewBucket(path)
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	metas, err := NewBlockQuerier(ctx, bucket).BlockMetas(ctx)
+	require.NoError(t, err)
+
+	for _, meta := range metas {
+		t.Run(fmt.Sprintf("block-v%d-%s", meta.Version, meta.ULID.String()), func(t *testing.T) {
+			q := NewSingleBlockQuerierFromMeta(ctx, bucket, meta)
+			require.NoError(t, q.Open(ctx))
+
+			profilesTypes, err := q.index.LabelValues("__profile_type__")
+			require.NoError(t, err)
+
+			profileCount := 0
+
+			for _, profileType := range profilesTypes {
+				t.Log(profileType)
+				profileTypeParts := strings.Split(profileType, ":")
+
+				it, err := q.SelectMatchingProfiles(ctx, &ingestv1.SelectProfilesRequest{
+					LabelSelector: "{}",
+					Start:         0,
+					End:           time.Now().UnixMilli(),
+					Type: &typesv1.ProfileType{
+						Name:       profileTypeParts[0],
+						SampleType: profileTypeParts[1],
+						SampleUnit: profileTypeParts[2],
+						PeriodType: profileTypeParts[3],
+						PeriodUnit: profileTypeParts[4],
+					},
+				})
+				require.NoError(t, err)
+
+				pcIt := &profileCounter{Iterator: it}
+
+				spanSelector, err := model.NewSpanSelector([]string{})
+				require.NoError(t, err)
+				resp, err := q.MergeBySpans(ctx, pcIt, spanSelector)
+				require.NoError(t, err)
+
+				require.Zero(t, resp.Total())
+				profileCount += pcIt.count
+			}
+
+			require.Zero(t, profileCount)
+		})
+	}
+}
+
 type fakeQuerier struct {
 	Querier
 	doErr bool

pkg/phlaredb/block_querier_test.go

@@ -160,16 +160,15 @@ func (q *headOnDiskQuerier) Series(ctx context.Context, params *ingestv1.SeriesR
 	return []*typesv1.Labels{}, nil
 }
 
-func (q *headOnDiskQuerier) MergeBySpans(_ context.Context, _ iter.Iterator[Profile], _ phlaremodel.SpanSelector) (*phlaremodel.Tree, error) {
-	//	sp, ctx := opentracing.StartSpanFromContext(ctx, "MergeBySpans")
-	//	defer sp.Finish()
-	//	r := symdb.NewResolver(ctx, q.head.symdb)
-	//	defer r.Release()
-	//	if err := mergeBySpans(ctx, q.rowGroup(), rows, r, spanSelector); err != nil {
-	//		return nil, err
-	//	}
-	//	return r.Tree()
-	return new(phlaremodel.Tree), nil
+func (q *headOnDiskQuerier) MergeBySpans(ctx context.Context, rows iter.Iterator[Profile], spanSelector phlaremodel.SpanSelector) (*phlaremodel.Tree, error) {
+	sp, ctx := opentracing.StartSpanFromContext(ctx, "MergeBySpans")
+	defer sp.Finish()
+	r := symdb.NewResolver(ctx, q.head.symdb)
+	defer r.Release()
+	if err := mergeBySpans(ctx, q.rowGroup(), rows, r, spanSelector); err != nil {
+		return nil, err
+	}
+	return r.Tree()
 }
 
 func (q *headOnDiskQuerier) Sort(in []Profile) []Profile {

pkg/phlaredb/head_queriers.go

@@ -3,6 +3,7 @@ package phlaredb
 import (
 	"context"
 	"sort"
+	"strings"
 
 	"github.com/google/pprof/profile"
 	"github.com/opentracing/opentracing-go"
@@ -54,16 +55,15 @@ func (b *singleBlockQuerier) MergeByLabels(ctx context.Context, rows iter.Iterat
 	return m.normalize(), nil
 }
 
-func (b *singleBlockQuerier) MergeBySpans(_ context.Context, _ iter.Iterator[Profile], _ phlaremodel.SpanSelector) (*phlaremodel.Tree, error) {
-	//	sp, _ := opentracing.StartSpanFromContext(ctx, "MergeBySpans - Block")
-	//	defer sp.Finish()
-	//	r := symdb.NewResolver(ctx, b.symbols)
-	//	defer r.Release()
-	//	if err := mergeBySpans(ctx, b.profiles.file, rows, r, spanSelector); err != nil {
-	//		return nil, err
-	//	}
-	//	return r.Tree()
-	return new(phlaremodel.Tree), nil
+func (b *singleBlockQuerier) MergeBySpans(ctx context.Context, rows iter.Iterator[Profile], spanSelector phlaremodel.SpanSelector) (*phlaremodel.Tree, error) {
+	sp, _ := opentracing.StartSpanFromContext(ctx, "MergeBySpans - Block")
+	defer sp.Finish()
+	r := symdb.NewResolver(ctx, b.symbols)
+	defer r.Release()
+	if err := mergeBySpans(ctx, b.profiles.file, rows, r, spanSelector); err != nil {
+		return nil, err
+	}
+	return r.Tree()
 }
 
 type Source interface {
@@ -94,6 +94,42 @@ func mergeByStacktraces(ctx context.Context, profileSource Source, rows iter.Ite
 	return it.Err()
 }
 
+func mergeBySpans(ctx context.Context, profileSource Source, rows iter.Iterator[Profile], r *symdb.Resolver, spanSelector phlaremodel.SpanSelector) error {
+	sp, ctx := opentracing.StartSpanFromContext(ctx, "mergeBySpans")
+	defer sp.Finish()
+	if _, found := profileSource.Schema().Lookup(strings.Split("Samples.list.element.SpanID", ".")...); !found {
+		return nil
+	}
+	// clone the rows to be able to iterate over them twice
+	multiRows, err := iter.CloneN(rows, 3)
+	if err != nil {
+		return err
+	}
+	it := query.NewMultiRepeatedPageIterator(
+		repeatedColumnIter(ctx, profileSource, "Samples.list.element.StacktraceID", multiRows[0]),
+		repeatedColumnIter(ctx, profileSource, "Samples.list.element.Value", multiRows[1]),
+		repeatedColumnIter(ctx, profileSource, "Samples.list.element.SpanID", multiRows[2]),
+	)
+	defer it.Close()
+	for it.Next() {
+		values := it.At().Values
+		p := r.Partition(it.At().Row.StacktracePartition())
+		stacktraces := values[0]
+		sampleValues := values[1]
+		spans := values[2]
+		for i := 0; i < len(stacktraces); i++ {
+			spanID := spans[i].Uint64()
+			if spanID == 0 {
+				continue
+			}
+			if _, ok := spanSelector[spanID]; ok {
+				p[uint32(stacktraces[i].Int64())] += sampleValues[i].Int64()
+			}
+		}
+	}
+	return it.Err()
+}
+
 type seriesByLabels map[string]*typesv1.Series
 
 func (m seriesByLabels) normalize() []*typesv1.Series {