Merge pull request #527 from getsentry/txiao/feat/handle-profile-chun…

…ks-in-regressed-endpoint

feat(statistical-detectors): Handle profile chunks in regressed endpoint

CHANGELOG.md

@@ -27,6 +27,7 @@
 - Forward SDK info for legacy profiles to Kafka. ([#515](https://github.com/getsentry/vroom/pull/515))
 - Add more metadata fields to Chunk Kafka message. ([#518](https://github.com/getsentry/vroom/pull/518))
 - Ingest Android profile chunks. ([#521](https://github.com/getsentry/vroom/pull/521))
+- Handle profile chunks in regressed endpoint. ([#527](https://github.com/getsentry/vroom/pull/527))
 
 **Bug Fixes**:
 

cmd/vroom/regressed.go

@@ -25,7 +25,7 @@ func (env *environment) postRegressed(w http.ResponseWriter, r *http.Request) {
 	for _, regressedFunction := range regressedFunctions {
 		s := sentry.StartSpan(ctx, "processing")
 		s.Description = "Generating occurrence for payload"
-		occurrence, err := occurrence.ProcessRegressedFunction(ctx, env.storage, regressedFunction)
+		occurrence, err := occurrence.ProcessRegressedFunction(ctx, env.storage, regressedFunction, readJobs)
 		s.Finish()
 		if err != nil {
 			hub.CaptureException(err)

internal/chunk/android.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/getsentry/vroom/internal/clientsdk"
 	"github.com/getsentry/vroom/internal/debugmeta"
+	"github.com/getsentry/vroom/internal/frame"
 	"github.com/getsentry/vroom/internal/nodetree"
 	"github.com/getsentry/vroom/internal/platform"
 	"github.com/getsentry/vroom/internal/profile"
@@ -112,5 +113,15 @@ func (c AndroidChunk) GetOptions() utils.Options {
 	return c.Options
 }
 
+func (c AndroidChunk) GetFrameWithFingerprint(target uint32) (frame.Frame, error) {
+	for _, m := range c.Profile.Methods {
+		f := m.Frame()
+		if f.Fingerprint() == target {
+			return f, nil
+		}
+	}
+	return frame.Frame{}, frame.ErrFrameNotFound
+}
+
 func (c *AndroidChunk) Normalize() {
 }

internal/chunk/chunk.go

@@ -3,6 +3,7 @@ package chunk
 import (
 	"fmt"
 
+	"github.com/getsentry/vroom/internal/frame"
 	"github.com/getsentry/vroom/internal/platform"
 	"github.com/getsentry/vroom/internal/utils"
 )
@@ -19,6 +20,7 @@ type (
 		GetRelease() string
 		GetRetentionDays() int
 		GetOptions() utils.Options
+		GetFrameWithFingerprint(uint32) (frame.Frame, error)
 
 		DurationMS() uint64
 		EndTimestamp() float64

internal/chunk/sample.go

@@ -252,3 +252,12 @@ func (c SampleChunk) GetOrganizationID() uint64 {
 func (c SampleChunk) GetOptions() utils.Options {
 	return c.Options
 }
+
+func (c SampleChunk) GetFrameWithFingerprint(target uint32) (frame.Frame, error) {
+	for _, f := range c.Profile.Frames {
+		if f.Fingerprint() == target {
+			return f, nil
+		}
+	}
+	return frame.Frame{}, frame.ErrFrameNotFound
+}

internal/frame/frame.go

@@ -3,6 +3,7 @@ package frame
 import (
 	"crypto/md5"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"hash"
 	"hash/fnv"
@@ -21,6 +22,8 @@ var (
 		"Sentry": {},
 		"hermes": {},
 	}
+
+	ErrFrameNotFound = errors.New("Unable to find matching frame")
 )
 
 type (

internal/nodetree/nodetree.go

@@ -269,21 +269,6 @@ func (n *Node) Close(timestamp uint64) {
 	}
 }
 
-func (n *Node) FindNodeByFingerprint(target uint32) *Node {
-	if n.Frame.Fingerprint() == target {
-		return n
-	}
-
-	for _, child := range n.Children {
-		node := child.FindNodeByFingerprint(target)
-		if node != nil {
-			return node
-		}
-	}
-
-	return nil
-}
-
 func isSymbolicatedFrame(f frame.Frame) bool {
 	// React-native case
 	if f.Platform == platform.JavaScript && f.IsReactNative {

internal/occurrence/occurrence.go

@@ -237,7 +237,6 @@ func FromRegressedFunction(
 		EvidenceData: map[string]interface{}{
 			"organization_id": regressed.OrganizationID,
 			"project_id":      regressed.ProjectID,
-			"profile_id":      regressed.ProfileID,
 
 			// frame info
 			"file":        f.File,

internal/occurrence/regressed_frame.go

@@ -3,132 +3,114 @@ package occurrence
 import (
 	"context"
 	"errors"
-	"sync"
 
 	"github.com/getsentry/sentry-go"
-	"github.com/getsentry/vroom/internal/nodetree"
+	"github.com/getsentry/vroom/internal/chunk"
+	"github.com/getsentry/vroom/internal/frame"
+	"github.com/getsentry/vroom/internal/platform"
 	"github.com/getsentry/vroom/internal/profile"
 	"github.com/getsentry/vroom/internal/storageutil"
+	"github.com/getsentry/vroom/internal/utils"
 	"gocloud.dev/blob"
 )
 
 type RegressedFunction struct {
-	OrganizationID           uint64  `json:"organization_id"`
-	ProjectID                uint64  `json:"project_id"`
-	ProfileID                string  `json:"profile_id"`
-	Fingerprint              uint32  `json:"fingerprint"`
-	AbsolutePercentageChange float64 `json:"absolute_percentage_change"`
-	AggregateRange1          float64 `json:"aggregate_range_1"`
-	AggregateRange2          float64 `json:"aggregate_range_2"`
-	Breakpoint               uint64  `json:"breakpoint"`
-	TrendDifference          float64 `json:"trend_difference"`
-	TrendPercentage          float64 `json:"trend_percentage"`
-	UnweightedPValue         float64 `json:"unweighted_p_value"`
-	UnweightedTValue         float64 `json:"unweighted_t_value"`
+	OrganizationID           uint64                `json:"organization_id"`
+	ProjectID                uint64                `json:"project_id"`
+	ProfileID                string                `json:"profile_id"`
+	Example                  utils.ExampleMetadata `json:"example"`
+	Fingerprint              uint32                `json:"fingerprint"`
+	AbsolutePercentageChange float64               `json:"absolute_percentage_change"`
+	AggregateRange1          float64               `json:"aggregate_range_1"`
+	AggregateRange2          float64               `json:"aggregate_range_2"`
+	Breakpoint               uint64                `json:"breakpoint"`
+	TrendDifference          float64               `json:"trend_difference"`
+	TrendPercentage          float64               `json:"trend_percentage"`
+	UnweightedPValue         float64               `json:"unweighted_p_value"`
+	UnweightedTValue         float64               `json:"unweighted_t_value"`
 }
 
 func ProcessRegressedFunction(
 	ctx context.Context,
 	profilesBucket *blob.Bucket,
 	regressedFunction RegressedFunction,
+	jobs chan storageutil.ReadJob,
 ) (*Occurrence, error) {
-	s := sentry.StartSpan(ctx, "profile.read")
-	s.Description = "Read profile from GCS"
-	var p profile.Profile
-	objectName := profile.StoragePath(
-		regressedFunction.OrganizationID,
-		regressedFunction.ProjectID,
-		regressedFunction.ProfileID,
-	)
-	err := storageutil.UnmarshalCompressed(ctx, profilesBucket, objectName, &p)
-	s.Finish()
-	if err != nil {
-		return nil, err
+	results := make(chan storageutil.ReadJobResult, 1)
+	defer close(results)
+
+	if regressedFunction.ProfileID != "" {
+		// For back compat, we should be use the example moving forwards
+		jobs <- profile.ReadJob{
+			Ctx:            ctx,
+			OrganizationID: regressedFunction.OrganizationID,
+			ProjectID:      regressedFunction.ProjectID,
+			ProfileID:      regressedFunction.ProfileID,
+			Storage:        profilesBucket,
+			Result:         results,
+		}
+	} else if regressedFunction.Example.ProfileID != "" {
+		jobs <- profile.ReadJob{
+			Ctx:            ctx,
+			OrganizationID: regressedFunction.OrganizationID,
+			ProjectID:      regressedFunction.ProjectID,
+			ProfileID:      regressedFunction.Example.ProfileID,
+			Storage:        profilesBucket,
+			Result:         results,
+		}
+	} else {
+		jobs <- chunk.ReadJob{
+			Ctx:            ctx,
+			OrganizationID: regressedFunction.OrganizationID,
+			ProjectID:      regressedFunction.ProjectID,
+			ProfilerID:     regressedFunction.Example.ProfilerID,
+			ChunkID:        regressedFunction.Example.ChunkID,
+			Storage:        profilesBucket,
+			Result:         results,
+		}
 	}
 
-	s = sentry.StartSpan(ctx, "processing")
-	s.Description = "Generate call trees"
-	calltreesByTID, err := p.CallTrees()
-	s.Finish()
-
+	res := <-results
+	platform, frame, err := getPlatformAndFrame(ctx, res, regressedFunction.Fingerprint)
 	if err != nil {
 		return nil, err
 	}
-
-	calltrees, exists := calltreesByTID[p.Transaction().ActiveThreadID]
-	if !exists {
-		return nil, errors.New("calltree not found")
-	}
-
-	s = sentry.StartSpan(ctx, "processing")
-	s.Description = "Searching for fingerprint"
-	var node *nodetree.Node
-	for _, calltree := range calltrees {
-		node = calltree.FindNodeByFingerprint(regressedFunction.Fingerprint)
-		if node != nil {
-			break
-		}
-	}
-	s.Finish()
-
-	if node == nil {
-		return nil, errors.New("fingerprint not found")
-	}
-
-	return FromRegressedFunction(p.Platform(), regressedFunction, node.Frame), nil
+	return FromRegressedFunction(platform, regressedFunction, frame), nil
 }
 
-func ProcessRegressedFunctions(
+func getPlatformAndFrame(
 	ctx context.Context,
-	hub *sentry.Hub,
-	profilesBucket *blob.Bucket,
-	regressedFunctions []RegressedFunction,
-	numWorkers int,
-) []*Occurrence {
-	if len(regressedFunctions) < numWorkers {
-		numWorkers = len(regressedFunctions)
-	}
-
-	var wg sync.WaitGroup
-	wg.Add(numWorkers)
-
-	regressedChan := make(chan RegressedFunction, numWorkers)
-	occurrenceChan := make(chan *Occurrence)
-
-	for i := 0; i < numWorkers; i++ {
-		go func() {
-			defer wg.Done()
-			for regressedFunction := range regressedChan {
-				occurrence, err := ProcessRegressedFunction(ctx, profilesBucket, regressedFunction)
-				if err != nil {
-					hub.CaptureException(err)
-					continue
-				} else if occurrence == nil {
-					continue
-				}
+	res storageutil.ReadJobResult,
+	target uint32,
+) (platform.Platform, frame.Frame, error) {
+	var platform platform.Platform
+	var frame frame.Frame
 
-				occurrenceChan <- occurrence
-			}
-		}()
+	err := res.Error()
+	if err != nil {
+		return platform, frame, err
 	}
 
-	go func() {
-		for _, regressedFunction := range regressedFunctions {
-			regressedChan <- regressedFunction
-		}
-		close(regressedChan)
-
-		// wait until all the profiles have been processed
-		// then we can close the occurrence channel and collect
-		// any occurrences that have been created
-		wg.Wait()
-		close(occurrenceChan)
-	}()
+	s := sentry.StartSpan(ctx, "processing")
+	s.Description = "Searching for fingerprint"
+	defer s.Finish()
 
-	occurrences := []*Occurrence{}
-	for occurrence := range occurrenceChan {
-		occurrences = append(occurrences, occurrence)
+	if result, ok := res.(profile.ReadJobResult); ok {
+		platform = result.Profile.Platform()
+		frame, err = result.Profile.GetFrameWithFingerprint(target)
+		if err != nil {
+			return platform, frame, err
+		}
+	} else if result, ok := res.(chunk.ReadJobResult); ok {
+		platform = result.Chunk.GetPlatform()
+		frame, err = result.Chunk.GetFrameWithFingerprint(target)
+		if err != nil {
+			return platform, frame, err
+		}
+	} else {
+		// This should never happen
+		return platform, frame, errors.New("unexpected result from storage")
 	}
 
-	return occurrences
+	return platform, frame, nil
 }

internal/profile/android.go

@@ -597,3 +597,14 @@ func (p Android) ActiveThreadID() uint64 {
 	}
 	return 0
 }
+
+func (p Android) GetFrameWithFingerprint(target uint32) (frame.Frame, error) {
+	for _, m := range p.Methods {
+		f := m.Frame()
+		if f.Fingerprint() == target {
+			return f, nil
+		}
+	}
+	// TODO: handle react native
+	return frame.Frame{}, frame.ErrFrameNotFound
+}

internal/profile/legacy.go

@@ -363,6 +363,10 @@ func (p LegacyProfile) GetOptions() utils.Options {
 	return p.Options
 }
 
+func (p LegacyProfile) GetFrameWithFingerprint(target uint32) (frame.Frame, error) {
+	return p.Trace.GetFrameWithFingerprint(target)
+}
+
 // This is to be used for ReactNative JS profile only since it works based on the
 // assumption that we'll only have 1 thread in the JS profile, as is the case
 // for ReactNative.

internal/profile/profile.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/getsentry/vroom/internal/debugmeta"
+	"github.com/getsentry/vroom/internal/frame"
 	"github.com/getsentry/vroom/internal/measurements"
 	"github.com/getsentry/vroom/internal/metadata"
 	"github.com/getsentry/vroom/internal/nodetree"
@@ -42,6 +43,7 @@ type (
 		IsSampled() bool
 		SetProfileID(ID string)
 		GetOptions() utils.Options
+		GetFrameWithFingerprint(uint32) (frame.Frame, error)
 	}
 
 	Profile struct {
@@ -173,3 +175,7 @@ func (p *Profile) Measurements() map[string]measurements.Measurement {
 func (p *Profile) GetOptions() utils.Options {
 	return p.profile.GetOptions()
 }
+
+func (p *Profile) GetFrameWithFingerprint(target uint32) (frame.Frame, error) {
+	return p.profile.GetFrameWithFingerprint(target)
+}

internal/profile/trace.go

@@ -1,6 +1,7 @@
 package profile
 
 import (
+	"github.com/getsentry/vroom/internal/frame"
 	"github.com/getsentry/vroom/internal/nodetree"
 	"github.com/getsentry/vroom/internal/speedscope"
 )
@@ -10,5 +11,6 @@ type (
 		ActiveThreadID() uint64
 		CallTrees() map[uint64][]*nodetree.Node
 		Speedscope() (speedscope.Output, error)
+		GetFrameWithFingerprint(uint32) (frame.Frame, error)
 	}
 )