Merge pull request #372 from parca-dev/release-0.2

Merge `release-0.2` back into `main`

pkg/query/query.go

@@ -111,6 +111,9 @@ func (q *Query) QueryRange(ctx context.Context, req *pb.QueryRangeRequest) (*pb.
 		it := series.Iterator()
 		for it.Next() {
 			p := it.At()
+			if p.ProfileMeta().Timestamp == 0 {
+				return nil, status.Error(codes.Internal, "profile's timestamp is 0")
+			}
 			metricsSeries.Samples = append(metricsSeries.Samples, &pb.MetricsSample{
 				Timestamp: timestamppb.New(timestamp.Time(p.ProfileMeta().Timestamp)),
 				Value:     p.ProfileTree().RootCumulativeValue(),

pkg/storage/flamegraph.go

@@ -167,11 +167,15 @@ func GenerateFlamegraph(
 			outerMost, innerMost := locationToTreeNodes(l, 0, 0)
 
 			flamegraphStack.Peek().node.Children = append(flamegraphStack.Peek().node.Children, outerMost)
-			flamegraphStack.Push(&TreeStackEntry{
-				node: innerMost,
-			})
-			if int32(len(flamegraphStack)) > flamegraph.Height {
-				flamegraph.Height = int32(len(flamegraphStack))
+
+			steppedInto := it.StepInto()
+			if steppedInto {
+				flamegraphStack.Push(&TreeStackEntry{
+					node: innerMost,
+				})
+				if int32(len(flamegraphStack)) > flamegraph.Height {
+					flamegraph.Height = int32(len(flamegraphStack))
+				}
 			}
 
 			for _, n := range child.FlatValues() {
@@ -191,7 +195,6 @@ func GenerateFlamegraph(
 				}
 			}
 
-			it.StepInto()
 			continue
 		}
 
@@ -301,6 +304,8 @@ func mergeChildren(node *pb.FlamegraphNode, compare, equals func(a, b *pb.Flameg
 		return compare(node.Children[i], node.Children[j])
 	})
 
+	var cumulative int64
+
 	i, j := 0, 1
 	for i < len(node.Children)-1 {
 		current, next := node.Children[i], node.Children[j]
@@ -311,6 +316,7 @@ func mergeChildren(node *pb.FlamegraphNode, compare, equals func(a, b *pb.Flameg
 				current.Meta.Mapping = &pb.Mapping{}
 			}
 
+			cumulative += next.Cumulative
 			current.Cumulative += next.Cumulative
 			current.Diff += next.Diff
 			current.Children = append(current.Children, next.Children...)
@@ -320,6 +326,11 @@ func mergeChildren(node *pb.FlamegraphNode, compare, equals func(a, b *pb.Flameg
 		}
 		i, j = i+1, j+1
 	}
+
+	// TODO: This is just a safeguard and should be properly fixed before this function.
+	if node.Cumulative < cumulative {
+		node.Cumulative = cumulative
+	}
 }
 
 func compareByName(a, b *pb.FlamegraphNode) bool {

pkg/storage/series_iterator.go

@@ -289,17 +289,21 @@ func (n *MemSeriesIteratorTreeNode) FlatValues() []*ProfileTreeValueNode {
 	return res
 }
 
-func getIndexRange(it MemSeriesValuesIterator, numSamples uint16, mint, maxt int64) (uint64, uint64, error) {
+func getIndexRange(it MemSeriesValuesIterator, numSamples uint64, mint, maxt int64) (uint64, uint64, error) {
 	// figure out the index of the first sample > mint and the last sample < maxt
 	start := uint64(0)
 	end := uint64(0)
-	i := uint16(0)
+	i := uint64(0)
 	for it.Next() {
 		if i == numSamples {
 			end++
 			break
 		}
 		t := it.At()
+		// MultiChunkIterator might return sparse values - shouldn't usually happen though.
+		if t == 0 {
+			break
+		}
 		if t < mint {
 			start++
 		}

pkg/storage/series_iterator_merge.go

@@ -44,15 +44,18 @@ func (ms *MemMergeSeries) Iterator() ProfileSeriesIterator {
 		maxt = ms.s.maxTime
 	}
 
+	var numSamples uint64
+
 	chunkStart, chunkEnd := ms.s.timestamps.indexRange(mint, maxt)
 	timestamps := make([]chunkenc.Chunk, 0, chunkEnd-chunkStart)
 	for _, t := range ms.s.timestamps[chunkStart:chunkEnd] {
+		numSamples += uint64(t.chunk.NumSamples())
 		timestamps = append(timestamps, t.chunk)
 	}
 
 	sl := &SliceProfileSeriesIterator{i: -1}
 
-	start, end, err := getIndexRange(NewMultiChunkIterator(timestamps), ms.s.numSamples, mint, maxt)
+	start, end, err := getIndexRange(NewMultiChunkIterator(timestamps), numSamples, mint, maxt)
 	if err != nil {
 		sl.err = err
 		return sl

pkg/storage/series_iterator_range.go

@@ -37,14 +37,17 @@ func (rs *MemRangeSeries) Iterator() ProfileSeriesIterator {
 	rs.s.mu.RLock()
 	defer rs.s.mu.RUnlock()
 
+	var numSamples uint64
+
 	chunkStart, chunkEnd := rs.s.timestamps.indexRange(rs.mint, rs.maxt)
 	timestamps := make([]chunkenc.Chunk, 0, chunkEnd-chunkStart)
 	for _, t := range rs.s.timestamps[chunkStart:chunkEnd] {
+		numSamples += uint64(t.chunk.NumSamples())
 		timestamps = append(timestamps, t.chunk)
 	}
 
 	it := NewMultiChunkIterator(timestamps)
-	start, end, err := getIndexRange(it, rs.s.numSamples, rs.mint, rs.maxt)
+	start, end, err := getIndexRange(it, numSamples, rs.mint, rs.maxt)
 	if err != nil {
 		return &MemRangeSeriesIterator{err: err}
 	}
@@ -111,7 +114,6 @@ func (rs *MemRangeSeries) Iterator() ProfileSeriesIterator {
 		periodsIterator.Seek(start)
 	}
 
-	numSamples := uint64(rs.s.numSamples)
 	if end-start < numSamples {
 		numSamples = end - start - 1
 	}

pkg/storage/series_iterator_root.go

@@ -35,14 +35,17 @@ func (rs *MemRootSeries) Iterator() ProfileSeriesIterator {
 	rs.s.mu.RLock()
 	defer rs.s.mu.RUnlock()
 
+	var numSamples uint64
+
 	chunkStart, chunkEnd := rs.s.timestamps.indexRange(rs.mint, rs.maxt)
 	timestamps := make([]chunkenc.Chunk, 0, chunkEnd-chunkStart)
 	for _, t := range rs.s.timestamps[chunkStart:chunkEnd] {
+		numSamples += uint64(t.chunk.NumSamples())
 		timestamps = append(timestamps, t.chunk)
 	}
 
 	it := NewMultiChunkIterator(timestamps)
-	start, end, err := getIndexRange(it, rs.s.numSamples, rs.mint, rs.maxt)
+	start, end, err := getIndexRange(it, numSamples, rs.mint, rs.maxt)
 	if start == end {
 		return &MemRootSeriesIterator{err: fmt.Errorf("no samples within the time range")}
 	}
@@ -58,7 +61,7 @@ func (rs *MemRootSeries) Iterator() ProfileSeriesIterator {
 		rootIterator.Seek(start)
 	}
 
-	numSamples := uint64(rs.s.numSamples)
+	// Set numSamples correctly if only subset selected.
 	if end-start < numSamples {
 		numSamples = end - start - 1
 	}

pkg/storage/series_iterator_test.go

@@ -269,6 +269,11 @@ func TestGetIndexRange(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, uint64(2), start)
 	require.Equal(t, uint64(4), end)
+
+	start, end, err = getIndexRange(NewMultiChunkIterator([]chunkenc.Chunk{c}), 123, 1, 12)
+	require.NoError(t, err)
+	require.Equal(t, uint64(0), start)
+	require.Equal(t, uint64(5), end)
 }
 
 func TestIteratorRangeSum(t *testing.T) {