Skip to content

Commit

Permalink
Add stats method to chunk metas
Browse files Browse the repository at this point in the history
  • Loading branch information
@jeschkies
jeschkies committed Jan 22, 2024
1 parent 86ca0fc commit c17a0f5
Show file tree
Hide file tree
Showing 11 changed files with 101 additions and 1,357 deletions.
79 changes: 23 additions & 56 deletions pkg/storage/stores/shipper/indexshipper/tsdb/index/chunk.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@ import (
"sort"

"github.com/prometheus/common/model"
"github.com/rdleal/intervalst/interval"

"github.com/grafana/loki/pkg/util"
"github.com/grafana/loki/pkg/util/encoding"
Expand Down Expand Up @@ -140,6 +139,25 @@ func (c ChunkMetas) Finalize() ChunkMetas {

}

// Stats returns the ChunkStats. It tries to accommodate for overlapping chunks.
func (c ChunkMetas) Stats(from, through int64) ChunkStats {
sort.Sort(c)

res := ChunkStats{}

// TODO: deduplicate

for _, chk := range c {
if overlap(from, through, chk.MinTime, chk.MaxTime) {
res.AddChunk(&chk, from, through)
} else if chk.MinTime >= through {
break
}
}

return res
}

// Add adds ChunkMeta at the right place in order. It assumes existing ChunkMetas have already been sorted by using Finalize.
// There is no chance of a data loss even if the chunks are not sorted because the chunk would anyways be added so the assumption is relatively safe.
func (c ChunkMetas) Add(chk ChunkMeta) ChunkMetas {
Expand Down Expand Up @@ -271,7 +289,6 @@ type chunkPageMarkers []chunkPageMarker

type ChunkStats struct {
Chunks, KB, Entries uint64
i *interval.MultiValueSearchTree[*ChunkMeta, int64]
}

func (cs *ChunkStats) addRaw(chunks int, kb, entries uint32) {
Expand All @@ -280,60 +297,10 @@ func (cs *ChunkStats) addRaw(chunks int, kb, entries uint32) {
cs.Entries += uint64(entries)
}

func (cs *ChunkStats) AddChunkMarker(marker chunkPageMarker) {
if cs.i == nil {
cs.i = interval.NewMultiValueSearchTree[*ChunkMeta, int64](cmp)
}

chk := &ChunkMeta{
MinTime: marker.MinTime,
MaxTime: marker.MaxTime,
KB: marker.KB,
Entries: marker.Entries,
}
rest := []*ChunkMeta{chk}
if intersections, ok := cs.i.AllIntersections(marker.MinTime, marker.MaxTime); ok {
for _, i := range intersections {
newRest := make([]*ChunkMeta, 0)
for _, r := range rest {
newRest = append(newRest, Sub(r, i)...)
}
rest = newRest
}
}

for _, r := range rest {
cs.i.Insert(r.MinTime, r.MaxTime, r) // nolint:errcheck
cs.addRaw(marker.ChunksInPage, r.KB, r.Entries)
}
}

func (cs *ChunkStats) AddChunk(chk *ChunkMeta, from, through int64) {
if cs.i == nil {
cs.i = interval.NewMultiValueSearchTree[*ChunkMeta, int64](cmp)
}

rest := []*ChunkMeta{chk}
if intersections, ok := cs.i.AllIntersections(chk.MinTime, chk.MaxTime); ok {
for _, i := range intersections {
newRest := make([]*ChunkMeta, 0)
for _, r := range rest {
newRest = append(newRest, Sub(r, i)...)
}
rest = newRest
}
}

for _, chk := range rest {
cs.i.Insert(chk.MinTime, chk.MaxTime, chk) // nolint:errcheck
factor := util.GetFactorOfTime(from, through, chk.MinTime, chk.MaxTime)
kb := uint32(float64(chk.KB) * factor)
entries := uint32(float64(chk.Entries) * factor)

cs.addRaw(1, kb, entries)
}
}
factor := util.GetFactorOfTime(from, through, chk.MinTime, chk.MaxTime)
kb := uint32(float64(chk.KB) * factor)
entries := uint32(float64(chk.Entries) * factor)

var cmp = func(t1, t2 int64) int {
return int(t1 - t2)
cs.addRaw(1, kb, entries)
}
144 changes: 69 additions & 75 deletions pkg/storage/stores/shipper/indexshipper/tsdb/index/chunk_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -603,7 +603,7 @@ func TestSearchWithPageMarkers(t *testing.T) {
func TestDecoderChunkStats(t *testing.T) {
for _, pageSize := range []int{2, 10} {
for _, version := range []int{
//FormatV2,
FormatV2,
FormatV3,
} {
for _, tc := range []struct {
Expand All @@ -612,94 +612,89 @@ func TestDecoderChunkStats(t *testing.T) {
from, through int64
exp ChunkStats
}{
/*
{
desc: "full range",
chks: []ChunkMeta{
{MinTime: 0, MaxTime: 1, KB: 1, Entries: 1},
{MinTime: 1, MaxTime: 2, KB: 1, Entries: 1},
{MinTime: 2, MaxTime: 3, KB: 1, Entries: 1},
{MinTime: 3, MaxTime: 4, KB: 1, Entries: 1},
{MinTime: 4, MaxTime: 5, KB: 1, Entries: 1},
},
from: 0,
through: 5,
exp: ChunkStats{
Chunks: 5,
KB: 5,
Entries: 5,
},
{
desc: "full range",
chks: []ChunkMeta{
{MinTime: 0, MaxTime: 1, KB: 1, Entries: 1},
{MinTime: 1, MaxTime: 2, KB: 1, Entries: 1},
{MinTime: 2, MaxTime: 3, KB: 1, Entries: 1},
{MinTime: 3, MaxTime: 4, KB: 1, Entries: 1},
{MinTime: 4, MaxTime: 5, KB: 1, Entries: 1},
},
*/
from: 0,
through: 5,
exp: ChunkStats{
Chunks: 5,
KB: 5,
Entries: 5,
},
},
{
desc: "overlapping",
chks: []ChunkMeta{
{MinTime: 0, MaxTime: 1, KB: 1, Entries: 1},
{MinTime: 1, MaxTime: 4, KB: 1, Entries: 1},
{MinTime: 2, MaxTime: 3, KB: 1, Entries: 1},
{MinTime: 3, MaxTime: 5, KB: 2, Entries: 2},
{MinTime: 3, MaxTime: 5, KB: 1, Entries: 1},
{MinTime: 4, MaxTime: 5, KB: 1, Entries: 1},
},
from: 0,
through: 5,
exp: ChunkStats{
Chunks: 4,
KB: 3,
Entries: 3,
Chunks: 5,
KB: 5,
Entries: 5,
},
},
/*
{
desc: "middle",
chks: []ChunkMeta{
{MinTime: 0, MaxTime: 1, KB: 1, Entries: 1},
{MinTime: 1, MaxTime: 2, KB: 1, Entries: 1},
{MinTime: 2, MaxTime: 3, KB: 1, Entries: 1},
{MinTime: 3, MaxTime: 4, KB: 1, Entries: 1},
{MinTime: 5, MaxTime: 6, KB: 1, Entries: 1},
},
from: 2,
through: 4,
exp: ChunkStats{
// technically the 2nd chunk overlaps, but we don't add
// any of it's stats as its only 1 nanosecond in the range
// and thus gets integer-divisioned to 0
Chunks: 3,
KB: 2,
Entries: 2,
},
{
desc: "middle",
chks: []ChunkMeta{
{MinTime: 0, MaxTime: 1, KB: 1, Entries: 1},
{MinTime: 1, MaxTime: 2, KB: 1, Entries: 1},
{MinTime: 2, MaxTime: 3, KB: 1, Entries: 1},
{MinTime: 3, MaxTime: 4, KB: 1, Entries: 1},
{MinTime: 5, MaxTime: 6, KB: 1, Entries: 1},
},
/*
{
desc: "middle with complete overlaps",
chks: []ChunkMeta{
// for example with pageSize=2
{MinTime: 0, MaxTime: 1, KB: 1, Entries: 1},
{MinTime: 1, MaxTime: 2, KB: 1, Entries: 1},
{MinTime: 2, MaxTime: 3, KB: 1, Entries: 1},
{MinTime: 3, MaxTime: 4, KB: 1, Entries: 1}, // partial overlap
{MinTime: 4, MaxTime: 5, KB: 1, Entries: 1}, // full overlap
{MinTime: 5, MaxTime: 6, KB: 1, Entries: 1},
{MinTime: 6, MaxTime: 7, KB: 1, Entries: 1}, // full overlap
{MinTime: 7, MaxTime: 8, KB: 1, Entries: 1},
{MinTime: 8, MaxTime: 9, KB: 1, Entries: 1}, // partial overlap
{MinTime: 9, MaxTime: 10, KB: 1, Entries: 1},
},
from: 4,
through: 9,
exp: ChunkStats{
// same deal with previous case, 1ns overlap isn't enough
// to include its data
Chunks: 6,
KB: 5,
Entries: 5,
},
from: 2,
through: 4,
exp: ChunkStats{
// technically the 2nd chunk overlaps, but we don't add
// any of it's stats as its only 1 nanosecond in the range
// and thus gets integer-divisioned to 0
Chunks: 3,
KB: 2,
Entries: 2,
},
*/
},
{
desc: "middle with complete overlaps",
chks: []ChunkMeta{
// for example with pageSize=2
{MinTime: 0, MaxTime: 1, KB: 1, Entries: 1},
{MinTime: 1, MaxTime: 2, KB: 1, Entries: 1},

{MinTime: 2, MaxTime: 3, KB: 1, Entries: 1},
{MinTime: 3, MaxTime: 4, KB: 1, Entries: 1}, // partial overlap

{MinTime: 4, MaxTime: 5, KB: 1, Entries: 1}, // full overlap
{MinTime: 5, MaxTime: 6, KB: 1, Entries: 1},

{MinTime: 6, MaxTime: 7, KB: 1, Entries: 1}, // full overlap
{MinTime: 7, MaxTime: 8, KB: 1, Entries: 1},

{MinTime: 8, MaxTime: 9, KB: 1, Entries: 1}, // partial overlap
{MinTime: 9, MaxTime: 10, KB: 1, Entries: 1},
},
from: 4,
through: 9,
exp: ChunkStats{
// same deal with previous case, 1ns overlap isn't enough
// to include its data
Chunks: 6,
KB: 5,
Entries: 5,
},
},
} {
t.Run(fmt.Sprintf("%s_version=%d_pageSize=%d", tc.desc, version, pageSize), func(t *testing.T) {
var w Writer
Expand All @@ -713,10 +708,9 @@ func TestDecoderChunkStats(t *testing.T) {

stats, err := dec.readChunkStats(version, &decbuf, 1, tc.from, tc.through)
require.Nil(t, err)

require.Equal(t, tc.exp.Chunks, stats.Chunks)
require.Equal(t, tc.exp.KB, stats.KB)
require.Equal(t, tc.exp.Entries, stats.Entries)
//require.Equal(t, tc.exp.Chunks, stats.Chunks)
})
}
}
Expand Down
38 changes: 9 additions & 29 deletions pkg/storage/stores/shipper/indexshipper/tsdb/index/index.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2290,13 +2290,14 @@ func (dec *Decoder) readChunkStatsV3(d *encoding.Decbuf, from, through int64) (r
// later used to incrementally skip pages
initialLn := d.Len()

chunkMetas := make(ChunkMetas, 0)
for markerIdx := 0; markerIdx < len(relevantPages); markerIdx++ {
curMarker := relevantPages[markerIdx]

if curMarker.subsetOf(from, through) {
// use aggregated stats for this page
res.AddChunkMarker(curMarker)
//res.addRaw(curMarker.ChunksInPage, curMarker.KB, curMarker.Entries)
chunkMetas = append(chunkMetas, ChunkMeta{MinTime: curMarker.MinTime, MaxTime: curMarker.MaxTime, KB: curMarker.KB, Entries: curMarker.Entries})
continue
}

Expand Down Expand Up @@ -2326,36 +2327,25 @@ func (dec *Decoder) readChunkStatsV3(d *encoding.Decbuf, from, through int64) (r

prevMaxT = chunkMeta.MaxTime

if overlap(from, through, chunkMeta.MinTime, chunkMeta.MaxTime) {
// add to stats
res.AddChunk(chunkMeta, from, through)
} else if chunkMeta.MinTime >= through {
break
}
chunkMetas = append(chunkMetas, *chunkMeta)
}
}

return res, d.Err()
return chunkMetas.Stats(from, through), d.Err()

}

func (dec *Decoder) accumulateChunkStats(d *encoding.Decbuf, nChunks int, from, through int64) (res ChunkStats, err error) {
var prevMaxT int64
chunkMeta := &ChunkMeta{}
for i := 0; i < nChunks; i++ {
chunkMetas := make(ChunkMetas, nChunks)
for i := range chunkMetas {
chunkMeta := &chunkMetas[i]
if err := readChunkMeta(d, prevMaxT, chunkMeta); err != nil {
return res, errors.Wrap(d.Err(), "read meta for chunk")
}
prevMaxT = chunkMeta.MaxTime

if overlap(from, through, chunkMeta.MinTime, chunkMeta.MaxTime) {
// add to stats
res.AddChunk(chunkMeta, from, through)
} else if chunkMeta.MinTime >= through {
break
}
}
return res, d.Err()
return chunkMetas.Stats(from, through), d.Err()
}

func (dec *Decoder) readChunkStatsPriorV3(d *encoding.Decbuf, seriesRef storage.SeriesRef, from, through int64) (res ChunkStats, err error) {
Expand All @@ -2367,17 +2357,7 @@ func (dec *Decoder) readChunkStatsPriorV3(d *encoding.Decbuf, seriesRef storage.
return ChunkStats{}, err
}

for _, chk := range chks {
if overlap(from, through, chk.MinTime, chk.MaxTime) {
res.AddChunk(&chk, from, through)
} else if chk.MinTime >= through {
break
}

}

return res, nil

return ChunkMetas(chks).Stats(from, through), nil
}

// Series decodes a series entry from the given byte slice into lset and chks.
Expand Down
9 changes: 0 additions & 9 deletions vendor/github.com/rdleal/intervalst/LICENSE
View file

This file was deleted.

Loading

0 comments on commit c17a0f5

Please sign in to comment.