feat: ingest chunk metrics (#495)

* ingest chunk metrics

CHANGELOG.md

@@ -21,6 +21,7 @@
 - Pass thread id to calltree generation ([#492](https://github.com/getsentry/vroom/pull/492))
 - Dual mode metrics endpoint ([#493](https://github.com/getsentry/vroom/pull/493))
 - Add optional generation of metrics during flamegraph aggregation ([#494](https://github.com/getsentry/vroom/pull/494))
+- Ingest function metrics from profile chunks ([#495](https://github.com/getsentry/vroom/pull/495))
 - Annotate flamegraph with profile data ([#501](https://github.com/getsentry/vroom/pull/501)), ([#502](https://github.com/getsentry/vroom/pull/502)), ([#503](https://github.com/getsentry/vroom/pull/503))
 
 **Bug Fixes**:

cmd/vroom/chunk.go

@@ -16,6 +16,7 @@ import (
 	"google.golang.org/api/googleapi"
 
 	"github.com/getsentry/vroom/internal/chunk"
+	"github.com/getsentry/vroom/internal/metrics"
 	"github.com/getsentry/vroom/internal/storageutil"
 )
 
@@ -113,6 +114,40 @@ func (env *environment) postChunk(w http.ResponseWriter, r *http.Request) {
 	}
 	s.Finish()
 
+	if c.Options.ProjectDSN != "" {
+		// nb.: here we don't have a specific thread ID, so we're going to ingest
+		// functions metrics from all the thread.
+		// That's ok as this data is not supposed to be transaction/span scoped,
+		// plus, we'll only retain application frames, so much of the system functions
+		// chaff will be dropped.
+		s = sentry.StartSpan(ctx, "processing")
+		callTrees, err := c.CallTrees(nil)
+		s.Finish()
+		if err != nil {
+			hub.CaptureException(err)
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+
+		s = sentry.StartSpan(ctx, "processing")
+		s.Description = "Extract functions"
+		functions := metrics.ExtractFunctionsFromCallTrees(callTrees)
+		functions = metrics.CapAndFilterFunctions(functions, maxUniqueFunctionsPerProfile, true)
+		s.Finish()
+
+		s = sentry.StartSpan(ctx, "processing")
+		s.Description = "Extract metrics from functions"
+		metrics := extractMetricsFromChunkFunctions(c, functions)
+		s.Finish()
+
+		if len(metrics) > 0 {
+			s = sentry.StartSpan(ctx, "processing")
+			s.Description = "Send functions metrics to generic metrics platform"
+			sendMetrics(ctx, c.Options.ProjectDSN, metrics, env.metricsClient)
+			s.Finish()
+		}
+	}
+
 	w.WriteHeader(http.StatusNoContent)
 }
 

cmd/vroom/profile.go

@@ -192,7 +192,7 @@ func (env *environment) postProfile(w http.ResponseWriter, r *http.Request) {
 			if len(metrics) > 0 {
 				s = sentry.StartSpan(ctx, "processing")
 				s.Description = "Send functions metrics to generic metrics platform"
-				sendMetrics(ctx, &p, metrics, env.metricsClient)
+				sendMetrics(ctx, p.GetOptions().ProjectDSN, metrics, env.metricsClient)
 				s.Finish()
 			}
 

cmd/vroom/utils.go

@@ -11,6 +11,7 @@ import (
 	"github.com/google/uuid"
 
 	"github.com/getsentry/sentry-go"
+	"github.com/getsentry/vroom/internal/chunk"
 	"github.com/getsentry/vroom/internal/nodetree"
 	"github.com/getsentry/vroom/internal/profile"
 )
@@ -40,6 +41,7 @@ func extractMetricsFromFunctions(p *profile.Profile, functions []nodetree.CallTr
 			"platform":       string(p.Platform()),
 			"environment":    p.Environment(),
 			"release":        p.Release(),
+			"profile_type":   "transaction",
 		}
 		duration := float64(function.SelfTimesNS[0] / 1e6)
 		summary := MetricSummary{
@@ -64,7 +66,7 @@ func extractMetricsFromFunctions(p *profile.Profile, functions []nodetree.CallTr
 	return metrics, metricsSummary
 }
 
-func sendMetrics(ctx context.Context, p *profile.Profile, metrics []sentry.Metric, mClient *http.Client) {
+func sendMetrics(ctx context.Context, dsn string, metrics []sentry.Metric, mClient *http.Client) {
 	id := strings.Replace(uuid.New().String(), "-", "", -1)
 	e := sentry.NewEvent()
 	e.EventID = sentry.EventID(id)
@@ -73,7 +75,7 @@ func sendMetrics(ctx context.Context, p *profile.Profile, metrics []sentry.Metri
 	tr := sentry.NewHTTPSyncTransport()
 	tr.Timeout = 5 * time.Second
 	tr.Configure(sentry.ClientOptions{
-		Dsn:           p.GetOptions().ProjectDSN,
+		Dsn:           dsn,
 		HTTPTransport: mClient.Transport,
 		HTTPClient:    mClient,
 	})
@@ -87,3 +89,33 @@ type MetricSummary struct {
 	Sum   float64
 	Count uint64
 }
+
+func extractMetricsFromChunkFunctions(c *chunk.Chunk, functions []nodetree.CallTreeFunction) []sentry.Metric {
+	metrics := make([]sentry.Metric, 0, len(functions))
+
+	for _, function := range functions {
+		if len(function.SelfTimesNS) == 0 {
+			continue
+		}
+		tags := map[string]string{
+			"project_id":     strconv.FormatUint(c.ProjectID, 10),
+			"fingerprint":    strconv.FormatUint(uint64(function.Fingerprint), 10),
+			"function":       function.Function,
+			"package":        function.Package,
+			"is_application": strconv.FormatBool(function.InApp),
+			"platform":       string(c.Platform),
+			"environment":    c.Environment,
+			"release":        c.Release,
+			"profile_type":   "continuous",
+		}
+		duration := float64(function.SelfTimesNS[0] / 1e6)
+		dm := sentry.NewDistributionMetric("profiles/function.duration", sentry.MilliSecond(), tags, int64(c.Received), duration)
+		// loop remaining selfTime durations
+		for i := 1; i < len(function.SelfTimesNS); i++ {
+			duration := float64(function.SelfTimesNS[i] / 1e6)
+			dm.Add(duration)
+		}
+		metrics = append(metrics, dm)
+	}
+	return metrics
+}

internal/flamegraph/flamegraph.go

@@ -578,24 +578,7 @@ func GetFlamegraphFromCandidates(
 			// if metrics aggregator is not null, while we're at it,
 			// compute the metrics as well
 			if ma != nil {
-				intChunkCallTrees := make(map[uint64][]*nodetree.Node)
-				var i uint64
-				for _, v := range chunkCallTrees {
-					// real TID here doesn't really matter as it's then
-					// discarded (not used) by ExtractFunctionsFromCallTrees.
-					// Here we're only assigning a random uint to make it compatible
-					// with ExtractFunctionsFromCallTrees which expects an
-					// uint64 -> []*nodetree.Node based on sample V1 int tid
-					// the TID.
-					//
-					// We could even refactor ExtractFunctionsFromCallTrees
-					// to simply accept []*nodetree.Node instead of a map
-					// but we'd end up moving the iteration from map to a slice
-					// somewhere else in the code.
-					intChunkCallTrees[i] = v
-					i++
-				}
-				functions := metrics.CapAndFilterFunctions(metrics.ExtractFunctionsFromCallTrees(intChunkCallTrees), int(ma.MaxUniqueFunctions), true)
+				functions := metrics.CapAndFilterFunctions(metrics.ExtractFunctionsFromCallTrees(chunkCallTrees), int(ma.MaxUniqueFunctions), true)
 				ma.AddFunctions(functions, example)
 			}
 		} else {

internal/flamegraph/flamegraph_test.go

@@ -302,6 +302,9 @@ func TestAnnotatingWithExamples(t *testing.T) {
 		cmpopts.SortSlices(func(a, b string) bool {
 			return a < b
 		}),
+		cmpopts.SortSlices(func(a, b int) bool {
+			return a < b
+		}),
 		// This option will order profile IDs since we only want to compare values and not order.
 		cmpopts.SortSlices(func(a, b utils.ExampleMetadata) bool {
 			if a.ProjectID != b.ProjectID {

internal/metrics/metrics.go

@@ -111,8 +111,8 @@ func quantile(values []uint64, q float64) (uint64, error) {
 	return values[index], nil
 }
 
-func ExtractFunctionsFromCallTrees(
-	callTrees map[uint64][]*nodetree.Node,
+func ExtractFunctionsFromCallTrees[T comparable](
+	callTrees map[T][]*nodetree.Node,
 ) []nodetree.CallTreeFunction {
 	functions := make(map[uint32]nodetree.CallTreeFunction, 0)
 
@@ -235,23 +235,7 @@ func (ma *Aggregator) GetMetricsFromCandidates(
 				hub.CaptureException(err)
 				continue
 			}
-			intChunkCallTrees := make(map[uint64][]*nodetree.Node)
-			var i uint64
-			for _, v := range chunkCallTrees {
-				// real TID here doesn't really matter as it's then
-				// discarded (not used) by ExtractFunctionsFromCallTrees.
-				// Here we're only assigning a random uint to make it compatible
-				// with ExtractFunctionsFromCallTrees which expects an
-				// uint64 -> []*nodetree.Node based on sample V1 int tid
-				// the TID.
-				//
-				// We could even refactor ExtractFunctionsFromCallTrees
-				// to simply accept []*nodetree.Node instead of a map
-				// but we'd end up moving the iteration from map to a slice
-				// somewhere else in the code.
-				intChunkCallTrees[i] = v
-				i++
-			}
+
 			resultMetadata = utils.NewExampleFromProfilerChunk(
 				result.Chunk.ProjectID,
 				result.Chunk.ProfilerID,
@@ -261,7 +245,7 @@ func (ma *Aggregator) GetMetricsFromCandidates(
 				result.Start,
 				result.End,
 			)
-			functions := CapAndFilterFunctions(ExtractFunctionsFromCallTrees(intChunkCallTrees), int(ma.MaxUniqueFunctions), true)
+			functions := CapAndFilterFunctions(ExtractFunctionsFromCallTrees(chunkCallTrees), int(ma.MaxUniqueFunctions), true)
 			ma.AddFunctions(functions, resultMetadata)
 		} else {
 			// this should never happen