feat: cumulative pprof merge for pull mode

pkg/adhoc/pull.go

@@ -50,7 +50,7 @@ func newPull(cfg *config.Adhoc, args []string, st *storage.Storage, logger *logr
 	}
 
 	p := parser.New(logger, st, e)
-	m := scrape.NewManager(logger, p, defaultMetricsRegistry)
+	m := scrape.NewManager(logger, p, defaultMetricsRegistry, true)
 	scrapeCfg := &(*scrapeconfig.DefaultConfig())
 	scrapeCfg.JobName = "adhoc"
 	scrapeCfg.EnabledProfiles = []string{"cpu", "mem"}

pkg/adhoc/push.go

@@ -36,7 +36,7 @@ func newPush(_ *config.Adhoc, args []string, st *storage.Storage, logger *logrus
 	p := parser.New(logger, st, e)
 	return push{
 		args:    args,
-		handler: server.NewIngestHandler(logger, p, func(*ingestion.IngestInput) {}, httputils.NewDefaultHelper(logger)),
+		handler: server.NewIngestHandler(logger, p, func(*ingestion.IngestInput) {}, httputils.NewDefaultHelper(logger), true),
 		logger:  logger,
 	}, nil
 }

pkg/cli/server.go

@@ -201,7 +201,8 @@ func newServerService(c *config.Server) (*serverService, error) {
 	svc.scrapeManager = scrape.NewManager(
 		svc.logger.WithField("component", "scrape-manager"),
 		ingester,
-		defaultMetricsRegistry)
+		defaultMetricsRegistry,
+		!svc.config.RemoteWrite.Enabled)
 
 	svc.controller, err = server.New(server.Config{
 		Configuration:           svc.config,

pkg/convert/pprof/profile.go

@@ -4,6 +4,8 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
+	"github.com/pyroscope-io/pyroscope/pkg/util/cumulativepprof"
 	"io"
 	"mime/multipart"
 	"sync"
@@ -22,6 +24,8 @@ type RawProfile struct {
 	// References the next profile in the sequence (cumulative type only).
 	next *RawProfile
 
+	mergers *cumulativepprof.Mergers
+
 	m sync.Mutex
 	// Initializes lazily on Bytes, if not present.
 	RawData             []byte // Represents raw request body as per ingestion API.
@@ -50,28 +54,73 @@ func (p *RawProfile) ContentType() string {
 // two consecutive samples to calculate the diff. If parser is not
 // present due to a failure, or sequence violation, the profiles will
 // be re-parsed.
-func (p *RawProfile) Push(profile []byte, cumulative bool) *RawProfile {
+func (p *RawProfile) Push(profile []byte, cumulative, mergeCumulative bool) *RawProfile {
 	p.m.Lock()
 	p.Profile = profile
 	p.RawData = nil
-	n := &RawProfile{
-		SampleTypeConfig: p.SampleTypeConfig,
-	}
 	if cumulative {
+		n := &RawProfile{
+			SampleTypeConfig: p.SampleTypeConfig,
+		}
 		// N.B the parser state is only propagated
 		// after successful Parse call.
 		n.PreviousProfile = p.Profile
 		p.next = n
+		if mergeCumulative {
+			mergers := p.mergers
+			if mergers == nil {
+				mergers = cumulativepprof.NewMergers()
+			}
+			err := p.mergeCumulativeLocked(mergers)
+			if err == nil {
+				n.mergers = mergers
+			}
+		}
 	}
 	p.m.Unlock()
 	return p.next
 }
 
+func (p *RawProfile) MergeCumulative(ms *cumulativepprof.Mergers) error {
+	p.m.Lock()
+	defer p.m.Unlock()
+	return p.mergeCumulativeLocked(ms)
+}
+
+func (p *RawProfile) mergeCumulativeLocked(ms *cumulativepprof.Mergers) error {
+	if p.Profile == nil && p.PreviousProfile == nil && p.RawData != nil && p.FormDataContentType != "" {
+		err := p.loadPprofFromForm()
+		if err != nil {
+			return err
+		}
+	}
+	if p.PreviousProfile == nil {
+		return ErrCumulativeMergeNoPreviousProfile
+	}
+	merged, stConfig, err := ms.Merge(p.PreviousProfile, p.Profile, p.SampleTypeConfig)
+	if err != nil {
+		return err
+	}
+	var mergedProfileBytes bytes.Buffer
+	err = merged.Write(&mergedProfileBytes)
+	if err != nil {
+		return err
+	}
+	p.Profile = mergedProfileBytes.Bytes()
+	p.PreviousProfile = nil
+	p.SampleTypeConfig = stConfig
+	p.RawData = nil
+	return nil
+}
+
 const (
 	formFieldProfile, formFileProfile                   = "profile", "profile.pprof"
 	formFieldPreviousProfile, formFilePreviousProfile   = "prev_profile", "profile.pprof"
 	formFieldSampleTypeConfig, formFileSampleTypeConfig = "sample_type_config", "sample_type_config.json"
 )
+var (
+	ErrCumulativeMergeNoPreviousProfile = errors.New("no previous profile for cumulative merge")
+)
 
 func (p *RawProfile) Bytes() ([]byte, error) {
 	p.m.Lock()

pkg/scrape/config/config.go

@@ -88,33 +88,33 @@ func DefaultConfig() *Config {
 			},
 			"mutex": {
 				Path:   "/debug/pprof/mutex",
-				Params: nil,
+				Params: url.Values{
+					"seconds": []string{"10"},
+				},
 				SampleTypes: map[string]*profile.SampleTypeConfig{
 					"contentions": {
 						DisplayName: "mutex_count",
 						Units:       metadata.LockSamplesUnits,
-						Cumulative:  true,
 					},
 					"delay": {
 						DisplayName: "mutex_duration",
 						Units:       metadata.LockNanosecondsUnits,
-						Cumulative:  true,
 					},
 				},
 			},
 			"block": {
 				Path:   "/debug/pprof/block",
-				Params: nil,
+				Params: url.Values{
+					"seconds": []string{"10"},
+				},
 				SampleTypes: map[string]*profile.SampleTypeConfig{
 					"contentions": {
 						DisplayName: "block_count",
 						Units:       metadata.LockSamplesUnits,
-						Cumulative:  true,
 					},
 					"delay": {
 						DisplayName: "block_duration",
 						Units:       metadata.LockNanosecondsUnits,
-						Cumulative:  true,
 					},
 				},
 			},

pkg/scrape/manager.go

@@ -45,19 +45,22 @@ type Manager struct {
 	targetSets    map[string][]*targetgroup.Group
 
 	reloadC chan struct{}
+
+	disableCumulativeMerge bool
 }
 
 // NewManager is the Manager constructor
-func NewManager(logger logrus.FieldLogger, p ingestion.Ingester, r prometheus.Registerer) *Manager {
+func NewManager(logger logrus.FieldLogger, p ingestion.Ingester, r prometheus.Registerer, disableCumulativeMerge bool) *Manager {
 	c := make(map[string]*config.Config)
 	return &Manager{
-		ingester:      p,
-		logger:        logger,
-		scrapeConfigs: c,
-		scrapePools:   make(map[string]*scrapePool),
-		stop:          make(chan struct{}),
-		reloadC:       make(chan struct{}, 1),
-		metrics:       newMetrics(r),
+		ingester:               p,
+		logger:                 logger,
+		scrapeConfigs:          c,
+		scrapePools:            make(map[string]*scrapePool),
+		stop:                   make(chan struct{}),
+		reloadC:                make(chan struct{}, 1),
+		metrics:                newMetrics(r),
+		disableCumulativeMerge: disableCumulativeMerge,
 	}
 }
 
@@ -90,7 +93,7 @@ func (m *Manager) reload() {
 					Errorf("reloading target set")
 				continue
 			}
-			sp, err := newScrapePool(scrapeConfig, m.ingester, m.logger, m.metrics)
+			sp, err := newScrapePool(scrapeConfig, m.ingester, m.logger, m.metrics, m.disableCumulativeMerge)
 			if err != nil {
 				m.logger.WithError(err).
 					WithField("scrape_pool", setName).

pkg/scrape/scrape.go

@@ -65,9 +65,11 @@ type scrapePool struct {
 	// set of hashes.
 	activeTargets  map[uint64]*Target
 	droppedTargets []*Target
+
+	disableCumulativeMerge bool
 }
 
-func newScrapePool(cfg *config.Config, p ingestion.Ingester, logger logrus.FieldLogger, m *metrics) (*scrapePool, error) {
+func newScrapePool(cfg *config.Config, p ingestion.Ingester, logger logrus.FieldLogger, m *metrics, disableCumulativeMerge bool) (*scrapePool, error) {
 	m.pools.Inc()
 	client, err := config.NewClientFromConfig(cfg.HTTPClientConfig, cfg.JobName)
 	if err != nil {
@@ -88,6 +90,7 @@ func newScrapePool(cfg *config.Config, p ingestion.Ingester, logger logrus.Field
 
 		metrics:     m,
 		poolMetrics: m.poolMetrics(cfg.JobName),
+		disableCumulativeMerge: disableCumulativeMerge,
 	}
 
 	return &sp, nil
@@ -105,6 +108,8 @@ func (sp *scrapePool) newScrapeLoop(s *scraper, i, t time.Duration) *scrapeLoop
 		delta:       d,
 		interval:    i,
 		timeout:     t,
+
+		disableCumulativeMerge: sp.disableCumulativeMerge,
 	}
 	x.ctx, x.cancel = context.WithCancel(sp.ctx)
 	return &x
@@ -324,6 +329,8 @@ type scrapeLoop struct {
 	delta    time.Duration
 	interval time.Duration
 	timeout  time.Duration
+
+	disableCumulativeMerge bool
 }
 
 func (sl *scrapeLoop) run() {
@@ -421,7 +428,8 @@ func (sl *scrapeLoop) scrape(startTime, endTime time.Time) error {
 	}
 
 	profile := sl.scraper.profile
-	sl.scraper.profile = profile.Push(buf.Bytes(), sl.scraper.cumulative)
+	sl.scraper.profile = profile.Push(buf.Bytes(), sl.scraper.cumulative, !sl.disableCumulativeMerge)
+
 	return sl.scraper.ingester.Ingest(ctx, &ingestion.IngestInput{
 		Profile: profile,
 		Metadata: ingestion.Metadata{

pkg/server/ingest.go

@@ -9,6 +9,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/pyroscope-io/pyroscope/pkg/util/cumulativepprof"
+
 	"github.com/pyroscope-io/pyroscope/pkg/convert/speedscope"
 	"github.com/sirupsen/logrus"
 
@@ -28,22 +30,25 @@ type ingestHandler struct {
 	ingester  ingestion.Ingester
 	onSuccess func(*ingestion.IngestInput)
 	httpUtils httputils.Utils
+
+	disableCumulativeMerge bool
 }
 
 func (ctrl *Controller) ingestHandler() http.Handler {
 	return NewIngestHandler(ctrl.log, ctrl.ingestser, func(pi *ingestion.IngestInput) {
 		ctrl.StatsInc("ingest")
 		ctrl.StatsInc("ingest:" + pi.Metadata.SpyName)
 		ctrl.appStats.Add(hashString(pi.Metadata.Key.AppName()))
-	}, ctrl.httpUtils)
+	}, ctrl.httpUtils, !ctrl.config.RemoteWrite.Enabled)
 }
 
-func NewIngestHandler(log *logrus.Logger, p ingestion.Ingester, onSuccess func(*ingestion.IngestInput), httpUtils httputils.Utils) http.Handler {
+func NewIngestHandler(log *logrus.Logger, p ingestion.Ingester, onSuccess func(*ingestion.IngestInput), httpUtils httputils.Utils, disableCumulativeMerge bool) http.Handler {
 	return ingestHandler{
-		log:       log,
-		ingester:  p,
-		onSuccess: onSuccess,
-		httpUtils: httpUtils,
+		log:                    log,
+		ingester:               p,
+		onSuccess:              onSuccess,
+		httpUtils:              httpUtils,
+		disableCumulativeMerge: disableCumulativeMerge,
 	}
 }
 
@@ -159,10 +164,14 @@ func (h ingestHandler) ingestInputFromRequest(r *http.Request) (*ingestion.Inges
 		}
 
 	case strings.Contains(contentType, "multipart/form-data"):
-		input.Profile = &pprof.RawProfile{
+		p := &pprof.RawProfile{
 			FormDataContentType: contentType,
 			RawData:             b,
 		}
+		if !h.disableCumulativeMerge {
+			p.MergeCumulative(cumulativepprof.NewMergers())
+		}
+		input.Profile = p
 	}
 
 	if input.Profile == nil {