hanzo/logs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: add support for fetching chunks with sizing info from the index to optimize GetShards calls (#19221)

Browse Source 
Signed-off-by: Sandeep Sukhani <sandeep.d.sukhani@gmail.com>
Co-authored-by: Christian Haudum <christian.haudum@gmail.com>
This commit is contained in:
Sandeep Sukhani
2025-10-09 08:10:18 +00:00
committed by GitHub
co-authored by Christian Haudum
parent 17b1418a92
commit a935a63abb
26 changed files with 335 additions and 307 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/indexgateway/gateway.go
+18 -78
View File
@@ -418,7 +418,7 @@ func (g *Gateway) GetShards(request *logproto.ShardsRequest, server logproto.Ind
return err
}
forSeries, ok := g.indexQuerier.HasForSeries(request.From, request.Through)
ok := g.indexQuerier.HasChunkSizingInfo(request.From, request.Through)
if !ok {
sp.AddEvent("index does not support forSeries", trace.WithAttributes(
attribute.String("action", "falling back to indexQuerier.GetShards impl"),
@@ -438,7 +438,7 @@ func (g *Gateway) GetShards(request *logproto.ShardsRequest, server logproto.Ind
return server.Send(shards)
}
return g.boundedShards(ctx, request, server, instanceID, p, forSeries)
return g.boundedShards(ctx, request, server, instanceID, p)
}
// boundedShards handles bounded shard requests, optionally returning precomputed chunks.
@@ -448,7 +448,6 @@ func (g *Gateway) boundedShards(
server logproto.IndexGateway_GetShardsServer,
instanceID string,
p chunk.Predicate,
forSeries sharding.ForSeries,
) error {
// TODO(owen-d): instead of using GetChunks which buffers _all_ the chunks
// (expensive when looking at the full fingerprint space), we should
@@ -467,27 +466,16 @@ func (g *Gateway) boundedShards(
defer sp.End()
// 1) for all bounds, get chunk refs
grps, _, err := g.indexQuerier.GetChunks(ctx, instanceID, req.From, req.Through, p, nil)
refs, err := g.indexQuerier.GetChunkRefsWithSizingInfo(ctx, instanceID, req.From, req.Through, p)
if err != nil {
return err
}
var ct int
for _, g := range grps {
ct += len(g)
}
ct := len(refs)
sp.AddEvent("queried local index", trace.WithAttributes(
attribute.Int("index_chunks_resolved", ct),
))
// TODO(owen-d): pool
refs := make([]*logproto.ChunkRef, 0, ct)
for _, cs := range grps {
for j := range cs {
refs = append(refs, &cs[j].ChunkRef)
}
}
filtered := refs
@@ -523,7 +511,7 @@ func (g *Gateway) boundedShards(
}
} else {
shards, chunkGrps, err := accumulateChunksToShards(ctx, instanceID, forSeries, req, p, filtered)
shards, chunkGrps, err := accumulateChunksToShards(req, refs)
if err != nil {
return err
}
@@ -597,71 +585,14 @@ func ExtractShardRequestMatchersAndAST(query string) (chunk.Predicate, error) {
}), nil
}
// TODO(owen-d): consider extending index impl to support returning chunkrefs _with_ sizing info
// TODO(owen-d): perf, this is expensive :(
func accumulateChunksToShards(
ctx context.Context,
user string,
forSeries sharding.ForSeries,
req *logproto.ShardsRequest,
p chunk.Predicate,
filtered []*logproto.ChunkRef,
filtered []logproto.ChunkRefWithSizingInfo,
) ([]logproto.Shard, []logproto.ChunkRefGroup, error) {
// map for looking up post-filtered chunks in O(n) while iterating the index again for sizing info
filteredM := make(map[model.Fingerprint][]refWithSizingInfo, 1024)
filteredM := make(map[model.Fingerprint][]logproto.ChunkRefWithSizingInfo, 1024)
for _, ref := range filtered {
x := refWithSizingInfo{ref: ref}
filteredM[model.Fingerprint(ref.Fingerprint)] = append(filteredM[model.Fingerprint(ref.Fingerprint)], x)
}
var mtx sync.Mutex
if err := forSeries.ForSeries(
ctx,
user,
v1.NewBounds(filtered[0].FingerprintModel(), filtered[len(filtered)-1].FingerprintModel()),
req.From, req.Through,
func(l labels.Labels, fp model.Fingerprint, chks []tsdb_index.ChunkMeta) (stop bool) {
mtx.Lock()
defer mtx.Unlock()
// check if this is a fingerprint we need
if _, ok := filteredM[fp]; !ok {
return false
}
filteredChks := filteredM[fp]
var j int
outer:
for i := range filteredChks {
for j < len(chks) {
switch filteredChks[i].Cmp(chks[j]) {
case iter.Less:
// this chunk is not in the queried index, continue checking other chunks
continue outer
case iter.Greater:
// next chunk in index but didn't pass filter; continue
j++
continue
case iter.Eq:
// a match; set the sizing info
filteredChks[i].KB = chks[j].KB
filteredChks[i].Entries = chks[j].Entries
j++
continue outer
}
}
// we've finished this index's chunks; no need to keep checking filtered chunks
break
}
return false
},
p.Matchers...,
); err != nil {
return nil, nil, err
filteredM[model.Fingerprint(ref.Fingerprint)] = append(filteredM[model.Fingerprint(ref.Fingerprint)], ref)
}
collectedSeries := sharding.SizedFPs(sharding.SizedFPsPool.Get(len(filteredM)))
@@ -689,13 +620,22 @@ func accumulateChunksToShards(
return filtered[i].Fingerprint > uint64(s.Bounds.Max)
})
chkGrps = append(chkGrps, logproto.ChunkRefGroup{
Refs: filtered[from:through],
Refs: refsWithSizingInfoToRefs(filtered[from:through]),
})
}
return shards, chkGrps, nil
}
func refsWithSizingInfoToRefs(refsWithSizingInfo []logproto.ChunkRefWithSizingInfo) []*logproto.ChunkRef {
refs := make([]*logproto.ChunkRef, 0, len(refsWithSizingInfo))
for _, refWithSizingInfo := range refsWithSizingInfo {
refs = append(refs, &refWithSizingInfo.ChunkRef)
}
return refs
}
type refWithSizingInfo struct {
ref *logproto.ChunkRef
KB uint32
pkg/indexgateway/gateway_test.go
+23 -112
View File
@@ -16,7 +16,6 @@ import (
v2 "github.com/grafana/loki/v3/pkg/iter/v2"
"github.com/grafana/loki/v3/pkg/logproto"
"github.com/grafana/loki/v3/pkg/storage/chunk"
"github.com/grafana/loki/v3/pkg/storage/config"
"github.com/grafana/loki/v3/pkg/storage/stores/series/index"
tsdb_index "github.com/grafana/loki/v3/pkg/storage/stores/shipper/indexshipper/tsdb/index"
@@ -426,135 +425,47 @@ func TestRefWithSizingInfo(t *testing.T) {
// TODO(owen-d): more testing for specific cases
func TestAccumulateChunksToShards(t *testing.T) {
// only check eq by checksum for convenience -- we're not testing the comparison function here
mkRef := func(fp model.Fingerprint, checksum uint32) *logproto.ChunkRef {
return &logproto.ChunkRef{
mkRef := func(fp model.Fingerprint, checksum uint32) logproto.ChunkRef {
return logproto.ChunkRef{
Fingerprint: uint64(fp),
Checksum: checksum,
}
}
sized := func(ref *logproto.ChunkRef, kb, entries uint32) refWithSizingInfo {
return refWithSizingInfo{
ref: ref,
KB: kb,
Entries: entries,
sized := func(ref logproto.ChunkRef, kb, entries uint32) logproto.ChunkRefWithSizingInfo {
return logproto.ChunkRefWithSizingInfo{
ChunkRef: ref,
KB: kb,
Entries: entries,
}
}
fsImpl := func(series [][]refWithSizingInfo) sharding.ForSeriesFunc {
return sharding.ForSeriesFunc(
func(
_ context.Context,
_ string,
_ tsdb_index.FingerprintFilter,
_, _ model.Time,
fn func(
_ labels.Labels,
fp model.Fingerprint,
chks []tsdb_index.ChunkMeta,
) (stop bool), _ ...*labels.Matcher) error {
for _, s := range series {
chks := []tsdb_index.ChunkMeta{}
for _, r := range s {
chks = append(chks, tsdb_index.ChunkMeta{
Checksum: r.ref.Checksum,
KB: r.KB,
Entries: r.Entries,
})
}
if stop := fn(labels.EmptyLabels(), s[0].ref.FingerprintModel(), chks); stop {
return nil
}
}
return nil
},
)
}
filtered := []*logproto.ChunkRef{
filtered := []logproto.ChunkRefWithSizingInfo{
// shard 0
mkRef(1, 0),
mkRef(1, 1),
mkRef(1, 2),
sized(mkRef(1, 0), 100, 1),
sized(mkRef(1, 1), 100, 1),
sized(mkRef(1, 2), 100, 1),
// shard 1
mkRef(2, 10),
mkRef(2, 20),
mkRef(2, 30),
sized(mkRef(2, 10), 100, 1),
sized(mkRef(2, 20), 100, 1),
sized(mkRef(2, 30), 100, 1),
// shard 2 split across multiple series
mkRef(3, 10),
mkRef(4, 10),
mkRef(4, 20),
sized(mkRef(3, 10), 50, 1),
sized(mkRef(4, 10), 30, 1),
sized(mkRef(4, 20), 30, 1),
// last shard contains leftovers + skip a few fps in between
mkRef(7, 10),
sized(mkRef(7, 10), 25, 1),
}
series := [][]refWithSizingInfo{
{
// first series creates one shard since a shard can't contain partial series.
// no chunks were filtered out
sized(mkRef(1, 0), 100, 1),
sized(mkRef(1, 1), 100, 1),
sized(mkRef(1, 2), 100, 1),
},
{
// second shard also contains one series, but this series has chunks filtered out.
sized(mkRef(2, 0), 100, 1), // filtered out
sized(mkRef(2, 10), 100, 1), // included
sized(mkRef(2, 11), 100, 1), // filtered out
sized(mkRef(2, 20), 100, 1), // included
sized(mkRef(2, 21), 100, 1), // filtered out
sized(mkRef(2, 30), 100, 1), // included
sized(mkRef(2, 31), 100, 1), // filtered out
},
shards, grps, err := accumulateChunksToShards(&logproto.ShardsRequest{
TargetBytesPerShard: 100 << 10,
}, filtered)
// third shard contains multiple series.
// combined they have 110kb, which is above the target of 100kb
// but closer than leaving the second series out which would create
// a shard with 50kb
{
// first series, 50kb
sized(mkRef(3, 10), 50, 1), // 50kb
sized(mkRef(3, 11), 50, 1), // 50kb, not included
},
{
// second series
sized(mkRef(4, 10), 30, 1), // 30kb
sized(mkRef(4, 11), 30, 1), // 30kb, not included
sized(mkRef(4, 20), 30, 1), // 30kb
},
// Fourth shard contains a single series with 25kb,
// but iterates over non-included fp(s) before it
{
// register a series in the index which is not included in the filtered list
sized(mkRef(6, 10), 100, 1), // not included
sized(mkRef(6, 11), 100, 1), // not included
},
{
// last shard contains leftovers
sized(mkRef(7, 10), 25, 1),
sized(mkRef(7, 11), 100, 1), // not included
},
}
shards, grps, err := accumulateChunksToShards(
context.Background(),
"",
fsImpl(series),
&logproto.ShardsRequest{
TargetBytesPerShard: 100 << 10,
},
chunk.NewPredicate(nil, nil), // we're not checking matcher injection here
filtered,
)
expectedChks := [][]*logproto.ChunkRef{
expectedChks := [][]logproto.ChunkRefWithSizingInfo{
filtered[0:3],
filtered[3:6],
filtered[6:9],
@@ -604,7 +515,7 @@ func TestAccumulateChunksToShards(t *testing.T) {
for i := range shards {
require.Equal(t, exp[i], shards[i], "invalid shard at index %d", i)
for j := range grps[i].Refs {
require.Equal(t, expectedChks[i][j], grps[i].Refs[j], "invalid chunk in grp %d at index %d", i, j)
require.Equal(t, &expectedChks[i][j].ChunkRef, grps[i].Refs[j], "invalid chunk in grp %d at index %d", i, j)
}
}
require.Equal(t, len(exp), len(shards))
pkg/ingester/flush_test.go
+8
View File
@@ -500,6 +500,14 @@ func (s *testStore) HasForSeries(_, _ model.Time) (sharding.ForSeries, bool) {
return nil, false
}
func (s *testStore) HasChunkSizingInfo(_, _ model.Time) bool {
return false
}
func (s *testStore) GetChunkRefsWithSizingInfo(_ context.Context, _ string, _, _ model.Time, _ chunk.Predicate) ([]logproto.ChunkRefWithSizingInfo, error) {
return nil, nil
}
func (s *testStore) GetSchemaConfigs() []config.PeriodConfig {
return defaultPeriodConfigs
}
pkg/ingester/ingester_test.go
+8
View File
@@ -515,6 +515,14 @@ func (s *mockStore) HasForSeries(_, _ model.Time) (sharding.ForSeries, bool) {
return nil, false
}
func (s *mockStore) HasChunkSizingInfo(_, _ model.Time) bool {
return false
}
func (s *mockStore) GetChunkRefsWithSizingInfo(_ context.Context, _ string, _, _ model.Time, _ chunk.Predicate) ([]logproto.ChunkRefWithSizingInfo, error) {
return nil, nil
}
func (s *mockStore) Volume(_ context.Context, _ string, _, _ model.Time, limit int32, _ []string, _ string, _ ...*labels.Matcher) (*logproto.VolumeResponse, error) {
return &logproto.VolumeResponse{
Volumes: []logproto.Volume{
pkg/logproto/types.go
+24
View File
@@ -7,3 +7,27 @@ import (
func (c *ChunkRef) FingerprintModel() model.Fingerprint {
return model.Fingerprint(c.Fingerprint)
}
type ChunkRefWithSizingInfo struct {
ChunkRef
KB uint32
Entries uint32
}
// Less Compares chunks by (Fp, From, Through, checksum)
// Assumes User is equivalent
func (c *ChunkRef) Less(x ChunkRef) bool {
if c.Fingerprint != x.Fingerprint {
return c.Fingerprint < x.Fingerprint
}
if c.From != x.From {
return c.From < x.From
}
if c.Through != x.Through {
return c.Through < x.Through
}
return c.Checksum < x.Checksum
}
pkg/querier/querier_mock_test.go
+12
View File
@@ -354,6 +354,14 @@ func (s *storeMock) HasForSeries(_, _ model.Time) (sharding.ForSeries, bool) {
return nil, false
}
func (s *storeMock) HasChunkSizingInfo(_, _ model.Time) bool {
return false
}
func (s *storeMock) GetChunkRefsWithSizingInfo(_ context.Context, _ string, _, _ model.Time, _ chunk.Predicate) ([]logproto.ChunkRefWithSizingInfo, error) {
return nil, nil
}
func (s *storeMock) Volume(ctx context.Context, userID string, from, through model.Time, _ int32, targetLabels []string, _ string, matchers ...*labels.Matcher) (*logproto.VolumeResponse, error) {
args := s.Called(ctx, userID, from, through, targetLabels, matchers)
return args.Get(0).(*logproto.VolumeResponse), args.Error(1)
@@ -729,6 +737,10 @@ func (q *querierMock) HasForSeries(_, _ model.Time) (sharding.ForSeries, bool) {
return nil, false
}
func (q *querierMock) HasChunkSizingInfo(_, _ model.Time) bool {
return false
}
func (q *querierMock) IndexShards(_ context.Context, _ *loghttp.RangeQuery, _ uint64) (*logproto.ShardsResponse, error) {
return nil, errors.New("unimplemented")
}
pkg/storage/stores/composite_store.go
+30
View File
@@ -296,6 +296,36 @@ func (c CompositeStore) HasForSeries(from, through model.Time) (sharding.ForSeri
return wrapped, true
}
func (c CompositeStore) HasChunkSizingInfo(from, through model.Time) bool {
allStoresHaveChunkSizingInfo := true
_ = c.forStores(context.Background(), from, through, func(_ context.Context, from, through model.Time, store Store) error {
if !store.HasChunkSizingInfo(from, through) {
allStoresHaveChunkSizingInfo = false
}
return nil
})
return allStoresHaveChunkSizingInfo
}
func (c CompositeStore) GetChunkRefsWithSizingInfo(ctx context.Context, userID string, from, through model.Time, predicate chunk.Predicate) ([]logproto.ChunkRefWithSizingInfo, error) {
var chunks []logproto.ChunkRefWithSizingInfo
err := c.forStores(ctx, from, through, func(innerCtx context.Context, innerFrom, innerThrough model.Time, store Store) error {
chks, err := store.GetChunkRefsWithSizingInfo(innerCtx, userID, innerFrom, innerThrough, predicate)
if err != nil {
return err
}
chunks = append(chunks, chks...)
return nil
})
if err != nil {
return nil, err
}
return chunks, nil
}
func (c CompositeStore) GetChunkFetcher(tm model.Time) *fetcher.Fetcher {
// find the schema with the lowest start _after_ tm
j := sort.Search(len(c.stores), func(j int) bool {
pkg/storage/stores/composite_store_entry.go
+19
View File
@@ -200,6 +200,25 @@ func (c *storeEntry) HasForSeries(from, through model.Time) (sharding.ForSeries,
return c.indexReader.HasForSeries(from, through)
}
func (c *storeEntry) HasChunkSizingInfo(from, through model.Time) bool {
return c.indexReader.HasChunkSizingInfo(from, through)
}
func (c *storeEntry) GetChunkRefsWithSizingInfo(ctx context.Context, userID string, from, through model.Time, predicate chunk.Predicate) ([]logproto.ChunkRefWithSizingInfo, error) {
if ctx.Err() != nil {
return nil, ctx.Err()
}
shortcut, err := c.validateQueryTimeRange(ctx, userID, &from, &through)
if err != nil {
return nil, err
} else if shortcut {
return nil, nil
}
return c.indexReader.GetChunkRefsWithSizingInfo(ctx, userID, from, through, predicate)
}
func (c *storeEntry) validateQueryTimeRange(ctx context.Context, userID string, from *model.Time, through *model.Time) (bool, error) {
//nolint:ineffassign,staticcheck //Leaving ctx even though we don't currently use it, we want to make it available for when we might need it and hopefully will ensure us using the correct context at that time
pkg/storage/stores/composite_store_test.go
+8
View File
@@ -69,6 +69,14 @@ func (m mockStore) HasForSeries(_, _ model.Time) (sharding.ForSeries, bool) {
return nil, false
}
func (m mockStore) HasChunkSizingInfo(_, _ model.Time) bool {
return false
}
func (m mockStore) GetChunkRefsWithSizingInfo(_ context.Context, _ string, _, _ model.Time, _ chunk.Predicate) ([]logproto.ChunkRefWithSizingInfo, error) {
return nil, nil
}
func (m mockStore) Stop() {}
func TestCompositeStore(t *testing.T) {
pkg/storage/stores/index/index.go
+24 -1
View File
@@ -42,13 +42,18 @@ type StatsReader interface {
// If the underlying index supports it, this will return the ForSeries interface
// which is used in bloom-filter accelerated sharding calculation optimization.
HasForSeries(from, through model.Time) (sharding.ForSeries, bool)
// HasChunkSizingInfo tells whether the index type for the given period supports listing chunks with their sizing info
HasChunkSizingInfo(from, through model.Time) bool
// GetChunkRefsWithSizingInfo should only be called after if HasChunkSizingInfo acknowledges that underlying index supports listing chunks with sizing info
GetChunkRefsWithSizingInfo(ctx context.Context, userID string, from, through model.Time, predicate chunk.Predicate) ([]logproto.ChunkRefWithSizingInfo, error)
}
type Reader interface {
BaseReader
StatsReader
GetChunkRefs(ctx context.Context, userID string, from, through model.Time, predicate chunk.Predicate) ([]logproto.ChunkRef, error)
Filterable
GetChunkRefs(ctx context.Context, userID string, from, through model.Time, predicate chunk.Predicate) ([]logproto.ChunkRef, error)
}
type Writer interface {
@@ -86,6 +91,20 @@ func (m MonitoredReaderWriter) GetChunkRefs(ctx context.Context, userID string,
return chunks, nil
}
func (m MonitoredReaderWriter) GetChunkRefsWithSizingInfo(ctx context.Context, userID string, from, through model.Time, predicate chunk.Predicate) ([]logproto.ChunkRefWithSizingInfo, error) {
var chunks []logproto.ChunkRefWithSizingInfo
if err := loki_instrument.TimeRequest(ctx, "chunk_refs_with_sizing_info", instrument.NewHistogramCollector(m.metrics.indexQueryLatency), instrument.ErrorCode, func(ctx context.Context) error {
var err error
chunks, err = m.rw.GetChunkRefsWithSizingInfo(ctx, userID, from, through, predicate)
return err
}); err != nil {
return nil, err
}
return chunks, nil
}
func (m MonitoredReaderWriter) GetSeries(ctx context.Context, userID string, from, through model.Time, matchers ...*labels.Matcher) ([]labels.Labels, error) {
var lbls []labels.Labels
if err := loki_instrument.TimeRequest(ctx, "series", instrument.NewHistogramCollector(m.metrics.indexQueryLatency), instrument.ErrorCode, func(ctx context.Context) error {
@@ -215,3 +234,7 @@ func (m MonitoredReaderWriter) HasForSeries(from, through model.Time) (sharding.
}
return nil, false
}
func (m MonitoredReaderWriter) HasChunkSizingInfo(from, through model.Time) bool {
return m.rw.HasChunkSizingInfo(from, through)
}
pkg/storage/stores/series/series_index_gateway_store.go
+8
View File
@@ -65,6 +65,10 @@ func (c *IndexGatewayClientStore) GetChunkRefs(ctx context.Context, _ string, fr
return result, nil
}
func (c *IndexGatewayClientStore) GetChunkRefsWithSizingInfo(_ context.Context, _ string, _, _ model.Time, _ chunk.Predicate) ([]logproto.ChunkRefWithSizingInfo, error) {
panic("store does not support getting chunk refs with sizing info")
}
func (c *IndexGatewayClientStore) GetSeries(ctx context.Context, _ string, from, through model.Time, matchers ...*labels.Matcher) ([]labels.Labels, error) {
resp, err := c.client.GetSeries(ctx, &logproto.GetSeriesRequest{
From: from,
@@ -153,3 +157,7 @@ func (c *IndexGatewayClientStore) IndexChunk(_ context.Context, _, _ model.Time,
func (c *IndexGatewayClientStore) HasForSeries(_, _ model.Time) (sharding.ForSeries, bool) {
return nil, false
}
func (c *IndexGatewayClientStore) HasChunkSizingInfo(_, _ model.Time) bool {
return false
}
pkg/storage/stores/series/series_index_store.go
+9
View File
@@ -198,6 +198,10 @@ func (c *IndexReaderWriter) GetChunkRefs(ctx context.Context, userID string, fro
return chunks, nil
}
func (c *IndexReaderWriter) GetChunkRefsWithSizingInfo(_ context.Context, _ string, _, _ model.Time, _ chunk.Predicate) ([]logproto.ChunkRefWithSizingInfo, error) {
panic("store does not support getting chunk refs with sizing info")
}
func (c *IndexReaderWriter) SetChunkFilterer(f chunk.RequestChunkFilterer) {
c.chunkFilterer = f
}
@@ -796,3 +800,8 @@ func (c *IndexReaderWriter) GetShards(
func (c *IndexReaderWriter) HasForSeries(_, _ model.Time) (sharding.ForSeries, bool) {
return nil, false
}
// old index stores do not have chunk sizing info
func (c *IndexReaderWriter) HasChunkSizingInfo(_, _ model.Time) bool {
return false
}
pkg/storage/stores/shipper/indexshipper/tsdb/head_manager.go
+2 -1
View File
@@ -24,6 +24,7 @@ import (
"github.com/prometheus/prometheus/util/compression"
"go.uber.org/atomic"
"github.com/grafana/loki/v3/pkg/logproto"
"github.com/grafana/loki/v3/pkg/storage/chunk"
"github.com/grafana/loki/v3/pkg/storage/chunk/client/util"
"github.com/grafana/loki/v3/pkg/storage/stores/shipper/indexshipper/tsdb/index"
@@ -769,7 +770,7 @@ func (t *tenantHeads) tenantIndex(userID string, from, through model.Time) (idx
}
func (t *tenantHeads) GetChunkRefs(ctx context.Context, userID string, from, through model.Time, _ []ChunkRef, fpFilter index.FingerprintFilter, matchers ...*labels.Matcher) ([]ChunkRef, error) {
func (t *tenantHeads) GetChunkRefs(ctx context.Context, userID string, from, through model.Time, _ []logproto.ChunkRefWithSizingInfo, fpFilter index.FingerprintFilter, matchers ...*labels.Matcher) ([]logproto.ChunkRefWithSizingInfo, error) {
idx, ok := t.tenantIndex(userID, from, through)
if !ok {
return nil, nil
pkg/storage/stores/shipper/indexshipper/tsdb/head_manager_test.go
+12 -8
View File
@@ -70,14 +70,18 @@ func (m *zeroValueLimits) DefaultLimits() *validation.Limits {
}
}
func chunkMetasToChunkRefs(user string, fp uint64, xs index.ChunkMetas) (res []ChunkRef) {
func chunkMetasToChunkRefs(user string, fp uint64, xs index.ChunkMetas) (res []logproto.ChunkRefWithSizingInfo) {
for _, x := range xs {
res = append(res, ChunkRef{
User: user,
Fingerprint: model.Fingerprint(fp),
Start: x.From(),
End: x.Through(),
Checksum: x.Checksum,
res = append(res, logproto.ChunkRefWithSizingInfo{
ChunkRef: logproto.ChunkRef{
UserID: user,
Fingerprint: fp,
From: x.From(),
Through: x.Through(),
Checksum: x.Checksum,
},
KB: x.KB,
Entries: x.Entries,
})
}
return
@@ -761,7 +765,7 @@ func BenchmarkTenantHeads(b *testing.B) {
wg.Add(1)
go func(r int) {
defer wg.Done()
var res []ChunkRef
var res []logproto.ChunkRefWithSizingInfo
tenant := r % nTenants
// nolint:ineffassign,staticcheck
pkg/storage/stores/shipper/indexshipper/tsdb/index.go
+3 -27
View File
@@ -6,6 +6,7 @@ import (
"github.com/prometheus/common/model"
"github.com/prometheus/prometheus/model/labels"
"github.com/grafana/loki/v3/pkg/logproto"
"github.com/grafana/loki/v3/pkg/storage/chunk"
"github.com/grafana/loki/v3/pkg/storage/stores/shipper/indexshipper/tsdb/index"
"github.com/grafana/loki/v3/pkg/storage/stores/shipper/indexshipper/tsdb/sharding"
@@ -16,31 +17,6 @@ type Series struct {
Fingerprint model.Fingerprint
}
type ChunkRef struct {
User string
Fingerprint model.Fingerprint
Start, End model.Time
Checksum uint32
}
// Compares by (Fp, Start, End, checksum)
// Assumes User is equivalent
func (r ChunkRef) Less(x ChunkRef) bool {
if r.Fingerprint != x.Fingerprint {
return r.Fingerprint < x.Fingerprint
}
if r.Start != x.Start {
return r.Start < x.Start
}
if r.End != x.End {
return r.End < x.End
}
return r.Checksum < x.Checksum
}
type shouldIncludeChunk func(index.ChunkMeta) bool
type Index interface {
@@ -58,7 +34,7 @@ type Index interface {
// the requested shard. If it is nil, TSDB will return all results,
// regardless of shard.
// Note: any shard used must be a valid factor of two, meaning `0_of_2` and `3_of_4` are fine, but `0_of_3` is not.
GetChunkRefs(ctx context.Context, userID string, from, through model.Time, res []ChunkRef, fpFilter index.FingerprintFilter, matchers ...*labels.Matcher) ([]ChunkRef, error)
GetChunkRefs(ctx context.Context, userID string, from, through model.Time, res []logproto.ChunkRefWithSizingInfo, fpFilter index.FingerprintFilter, matchers ...*labels.Matcher) ([]logproto.ChunkRefWithSizingInfo, error)
// Series follows the same semantics regarding the passed slice and shard as GetChunkRefs.
Series(ctx context.Context, userID string, from, through model.Time, res []Series, fpFilter index.FingerprintFilter, matchers ...*labels.Matcher) ([]Series, error)
LabelNames(ctx context.Context, userID string, from, through model.Time, matchers ...*labels.Matcher) ([]string, error)
@@ -71,7 +47,7 @@ type NoopIndex struct{}
func (NoopIndex) Close() error { return nil }
func (NoopIndex) Bounds() (_, through model.Time) { return }
func (NoopIndex) GetChunkRefs(_ context.Context, _ string, _, _ model.Time, _ []ChunkRef, _ index.FingerprintFilter, _ ...*labels.Matcher) ([]ChunkRef, error) {
func (NoopIndex) GetChunkRefs(_ context.Context, _ string, _, _ model.Time, _ []logproto.ChunkRefWithSizingInfo, _ index.FingerprintFilter, _ ...*labels.Matcher) ([]logproto.ChunkRefWithSizingInfo, error) {
return nil, nil
}
pkg/storage/stores/shipper/indexshipper/tsdb/index_client.go
+20 -10
View File
@@ -119,9 +119,6 @@ func cleanMatchers(matchers ...*labels.Matcher) ([]*labels.Matcher, index.Finger
return matchers, nil, nil
}
// TODO(owen-d): synchronize logproto.ChunkRef and tsdb.ChunkRef so we don't have to convert.
// They share almost the same fields, so we can add the missing `KB` field to the proto and then
// use that within the tsdb package.
func (c *IndexClient) GetChunkRefs(ctx context.Context, userID string, from, through model.Time, predicate chunk.Predicate) ([]logproto.ChunkRef, error) {
matchers, shard, err := cleanMatchers(predicate.Matchers...)
if err != nil {
@@ -136,18 +133,27 @@ func (c *IndexClient) GetChunkRefs(ctx context.Context, userID string, from, thr
refs := make([]logproto.ChunkRef, 0, len(chks))
for _, chk := range chks {
refs = append(refs, logproto.ChunkRef{
Fingerprint: uint64(chk.Fingerprint),
UserID: chk.User,
From: chk.Start,
Through: chk.End,
Checksum: chk.Checksum,
})
refs = append(refs, chk.ChunkRef)
}
return refs, err
}
func (c *IndexClient) GetChunkRefsWithSizingInfo(ctx context.Context, userID string, from, through model.Time, predicate chunk.Predicate) ([]logproto.ChunkRefWithSizingInfo, error) {
matchers, shard, err := cleanMatchers(predicate.Matchers...)
if err != nil {
return nil, err
}
// TODO(owen-d): use a pool to reduce allocs here
chks, err := c.idx.GetChunkRefs(ctx, userID, from, through, nil, shard, matchers...)
if err != nil {
return nil, err
}
return chks, err
}
func (c *IndexClient) GetSeries(ctx context.Context, userID string, from, through model.Time, matchers ...*labels.Matcher) ([]labels.Labels, error) {
matchers, shard, err := cleanMatchers(matchers...)
if err != nil {
@@ -352,3 +358,7 @@ func withoutNameLabel(matchers []*labels.Matcher) []*labels.Matcher {
func (c *IndexClient) HasForSeries(_, _ model.Time) (sharding.ForSeries, bool) {
return c.idx, true
}
func (c *IndexClient) HasChunkSizingInfo(_, _ model.Time) bool {
return true
}
pkg/storage/stores/shipper/indexshipper/tsdb/index_shipper_querier.go
+2 -1
View File
@@ -10,6 +10,7 @@ import (
"github.com/prometheus/common/model"
"github.com/prometheus/prometheus/model/labels"
"github.com/grafana/loki/v3/pkg/logproto"
"github.com/grafana/loki/v3/pkg/storage/chunk"
"github.com/grafana/loki/v3/pkg/storage/config"
shipperindex "github.com/grafana/loki/v3/pkg/storage/stores/shipper/indexshipper/index"
@@ -84,7 +85,7 @@ func (i *indexShipperQuerier) Close() error {
return nil
}
func (i *indexShipperQuerier) GetChunkRefs(ctx context.Context, userID string, from, through model.Time, res []ChunkRef, fpFilter tsdbindex.FingerprintFilter, matchers ...*labels.Matcher) ([]ChunkRef, error) {
func (i *indexShipperQuerier) GetChunkRefs(ctx context.Context, userID string, from, through model.Time, res []logproto.ChunkRefWithSizingInfo, fpFilter tsdbindex.FingerprintFilter, matchers ...*labels.Matcher) ([]logproto.ChunkRefWithSizingInfo, error) {
idx, err := i.indices(ctx, from, through, userID)
if err != nil {
return nil, err
pkg/storage/stores/shipper/indexshipper/tsdb/lazy_index.go
+2 -1
View File
@@ -6,6 +6,7 @@ import (
"github.com/prometheus/common/model"
"github.com/prometheus/prometheus/model/labels"
"github.com/grafana/loki/v3/pkg/logproto"
"github.com/grafana/loki/v3/pkg/storage/chunk"
"github.com/grafana/loki/v3/pkg/storage/stores/shipper/indexshipper/tsdb/index"
)
@@ -36,7 +37,7 @@ func (f LazyIndex) Close() error {
return i.Close()
}
func (f LazyIndex) GetChunkRefs(ctx context.Context, userID string, from, through model.Time, res []ChunkRef, fpFilter index.FingerprintFilter, matchers ...*labels.Matcher) ([]ChunkRef, error) {
func (f LazyIndex) GetChunkRefs(ctx context.Context, userID string, from, through model.Time, res []logproto.ChunkRefWithSizingInfo, fpFilter index.FingerprintFilter, matchers ...*labels.Matcher) ([]logproto.ChunkRefWithSizingInfo, error) {
i, err := f()
if err != nil {
return nil, err
pkg/storage/stores/shipper/indexshipper/tsdb/multi_file_index.go
+7 -6
View File
@@ -11,6 +11,7 @@ import (
"github.com/prometheus/prometheus/model/labels"
"golang.org/x/sync/errgroup"
"github.com/grafana/loki/v3/pkg/logproto"
"github.com/grafana/loki/v3/pkg/storage/chunk"
"github.com/grafana/loki/v3/pkg/storage/stores/shipper/indexshipper/tsdb/index"
)
@@ -132,15 +133,15 @@ func (i *MultiIndex) forMatchingIndices(ctx context.Context, from, through model
}
func (i *MultiIndex) GetChunkRefs(ctx context.Context, userID string, from, through model.Time, res []ChunkRef, fpFilter index.FingerprintFilter, matchers ...*labels.Matcher) ([]ChunkRef, error) {
acc := newResultAccumulator(func(xs [][]ChunkRef) ([]ChunkRef, error) {
func (i *MultiIndex) GetChunkRefs(ctx context.Context, userID string, from, through model.Time, res []logproto.ChunkRefWithSizingInfo, fpFilter index.FingerprintFilter, matchers ...*labels.Matcher) ([]logproto.ChunkRefWithSizingInfo, error) {
acc := newResultAccumulator(func(xs [][]logproto.ChunkRefWithSizingInfo) ([]logproto.ChunkRefWithSizingInfo, error) {
if res == nil {
res = ChunkRefsPool.Get()
}
res = res[:0]
// keep track of duplicates
seen := make(map[ChunkRef]struct{})
seen := make(map[logproto.ChunkRef]struct{})
// TODO(owen-d): Do this more efficiently,
// not all indices overlap each other
@@ -150,17 +151,17 @@ func (i *MultiIndex) GetChunkRefs(ctx context.Context, userID string, from, thro
g := group
for _, ref := range g {
_, ok := seen[ref]
_, ok := seen[ref.ChunkRef]
if ok {
continue
}
seen[ref] = struct{}{}
seen[ref.ChunkRef] = struct{}{}
res = append(res, ref)
}
ChunkRefsPool.Put(g)
}
sort.Slice(res, func(i, j int) bool { return res[i].Less(res[j]) })
sort.Slice(res, func(i, j int) bool { return res[i].ChunkRef.Less(res[j].ChunkRef) })
return res, nil
})
pkg/storage/stores/shipper/indexshipper/tsdb/multi_file_index_test.go
+30 -21
View File
@@ -8,6 +8,7 @@ import (
"github.com/prometheus/prometheus/model/labels"
"github.com/stretchr/testify/require"
"github.com/grafana/loki/v3/pkg/logproto"
"github.com/grafana/loki/v3/pkg/storage/stores/shipper/indexshipper/tsdb/index"
)
@@ -70,34 +71,42 @@ func TestMultiIndex(t *testing.T) {
refs, err := idx.GetChunkRefs(context.Background(), "fake", 2, 5, nil, nil, labels.MustNewMatcher(labels.MatchEqual, "foo", "bar"))
require.Nil(t, err)
expected := []ChunkRef{
expected := []logproto.ChunkRefWithSizingInfo{
{
User: "fake",
Fingerprint: model.Fingerprint(labels.StableHash(mustParseLabels(`{foo="bar"}`))),
Start: 0,
End: 3,
Checksum: 0,
ChunkRef: logproto.ChunkRef{
UserID: "fake",
Fingerprint: labels.StableHash(mustParseLabels(`{foo="bar"}`)),
From: 0,
Through: 3,
Checksum: 0,
},
},
{
User: "fake",
Fingerprint: model.Fingerprint(labels.StableHash(mustParseLabels(`{foo="bar"}`))),
Start: 1,
End: 4,
Checksum: 1,
ChunkRef: logproto.ChunkRef{
UserID: "fake",
Fingerprint: labels.StableHash(mustParseLabels(`{foo="bar"}`)),
From: 1,
Through: 4,
Checksum: 1,
},
},
{
User: "fake",
Fingerprint: model.Fingerprint(labels.StableHash(mustParseLabels(`{foo="bar"}`))),
Start: 2,
End: 5,
Checksum: 2,
ChunkRef: logproto.ChunkRef{
UserID: "fake",
Fingerprint: labels.StableHash(mustParseLabels(`{foo="bar"}`)),
From: 2,
Through: 5,
Checksum: 2,
},
},
{
User: "fake",
Fingerprint: model.Fingerprint(labels.StableHash(mustParseLabels(`{foo="bar", bazz="buzz"}`))),
Start: 1,
End: 10,
Checksum: 3,
ChunkRef: logproto.ChunkRef{
UserID: "fake",
Fingerprint: labels.StableHash(mustParseLabels(`{foo="bar", bazz="buzz"}`)),
From: 1,
Through: 10,
Checksum: 3,
},
},
}
require.Equal(t, expected, refs)
pkg/storage/stores/shipper/indexshipper/tsdb/multitenant.go
+2 -1
View File
@@ -7,6 +7,7 @@ import (
"github.com/prometheus/common/model"
"github.com/prometheus/prometheus/model/labels"
"github.com/grafana/loki/v3/pkg/logproto"
"github.com/grafana/loki/v3/pkg/storage/chunk"
"github.com/grafana/loki/v3/pkg/storage/stores/shipper/indexshipper/tsdb/index"
)
@@ -45,7 +46,7 @@ func (m *MultiTenantIndex) SetChunkFilterer(chunkFilter chunk.RequestChunkFilter
func (m *MultiTenantIndex) Close() error { return m.idx.Close() }
func (m *MultiTenantIndex) GetChunkRefs(ctx context.Context, userID string, from, through model.Time, res []ChunkRef, fpFilter index.FingerprintFilter, matchers ...*labels.Matcher) ([]ChunkRef, error) {
func (m *MultiTenantIndex) GetChunkRefs(ctx context.Context, userID string, from, through model.Time, res []logproto.ChunkRefWithSizingInfo, fpFilter index.FingerprintFilter, matchers ...*labels.Matcher) ([]logproto.ChunkRefWithSizingInfo, error) {
return m.idx.GetChunkRefs(ctx, userID, from, through, res, fpFilter, withTenantLabelMatcher(userID, matchers)...)
}
pkg/storage/stores/shipper/indexshipper/tsdb/pool.go
+5 -4
View File
@@ -3,6 +3,7 @@ package tsdb
import (
"sync"
"github.com/grafana/loki/v3/pkg/logproto"
"github.com/grafana/loki/v3/pkg/storage/stores/shipper/indexshipper/tsdb/index"
)
@@ -33,14 +34,14 @@ type PoolChunkRefs struct {
pool sync.Pool
}
func (p *PoolChunkRefs) Get() []ChunkRef {
func (p *PoolChunkRefs) Get() []logproto.ChunkRefWithSizingInfo {
if xs := p.pool.Get(); xs != nil {
return xs.([]ChunkRef)
return xs.([]logproto.ChunkRefWithSizingInfo)
}
return make([]ChunkRef, 0, 1<<10)
return make([]logproto.ChunkRefWithSizingInfo, 0, 1<<10)
}
func (p *PoolChunkRefs) Put(xs []ChunkRef) {
func (p *PoolChunkRefs) Put(xs []logproto.ChunkRefWithSizingInfo) {
xs = xs[:0]
//nolint:staticcheck
p.pool.Put(xs)
pkg/storage/stores/shipper/indexshipper/tsdb/single_file_index.go
+12 -8
View File
@@ -14,6 +14,7 @@ import (
"github.com/prometheus/common/model"
"github.com/prometheus/prometheus/model/labels"
"github.com/grafana/loki/v3/pkg/logproto"
"github.com/grafana/loki/v3/pkg/storage/chunk"
"github.com/grafana/loki/v3/pkg/storage/stores/index/seriesvolume"
shipperindex "github.com/grafana/loki/v3/pkg/storage/stores/shipper/indexshipper/index"
@@ -211,7 +212,7 @@ func (i *TSDBIndex) forPostings(
return fn(p)
}
func (i *TSDBIndex) GetChunkRefs(ctx context.Context, userID string, from, through model.Time, res []ChunkRef, fpFilter index.FingerprintFilter, matchers ...*labels.Matcher) ([]ChunkRef, error) {
func (i *TSDBIndex) GetChunkRefs(ctx context.Context, userID string, from, through model.Time, res []logproto.ChunkRefWithSizingInfo, fpFilter index.FingerprintFilter, matchers ...*labels.Matcher) ([]logproto.ChunkRefWithSizingInfo, error) {
if res == nil {
res = ChunkRefsPool.Get()
}
@@ -219,13 +220,16 @@ func (i *TSDBIndex) GetChunkRefs(ctx context.Context, userID string, from, throu
if err := i.ForSeries(ctx, "", fpFilter, from, through, func(_ labels.Labels, fp model.Fingerprint, chks []index.ChunkMeta) (stop bool) {
for _, chk := range chks {
res = append(res, ChunkRef{
User: userID, // assumed to be the same, will be enforced by caller.
Fingerprint: fp,
Start: chk.From(),
End: chk.Through(),
Checksum: chk.Checksum,
res = append(res, logproto.ChunkRefWithSizingInfo{
ChunkRef: logproto.ChunkRef{
UserID: userID, // assumed to be the same, will be enforced by caller.
Fingerprint: uint64(fp),
From: chk.From(),
Through: chk.Through(),
Checksum: chk.Checksum,
},
KB: chk.KB,
Entries: chk.Entries,
})
}
return false
pkg/storage/stores/shipper/indexshipper/tsdb/single_file_index_test.go
+37 -27
View File
@@ -92,34 +92,42 @@ func TestSingleIdx(t *testing.T) {
refs, err := idx.GetChunkRefs(context.Background(), "fake", 1, 5, nil, nil, labels.MustNewMatcher(labels.MatchEqual, "foo", "bar"))
require.Nil(t, err)
expected := []ChunkRef{
expected := []logproto.ChunkRefWithSizingInfo{
{
User: "fake",
Fingerprint: model.Fingerprint(labels.StableHash(mustParseLabels(`{foo="bar"}`))),
Start: 0,
End: 3,
Checksum: 0,
ChunkRef: logproto.ChunkRef{
UserID: "fake",
Fingerprint: labels.StableHash(mustParseLabels(`{foo="bar"}`)),
From: 0,
Through: 3,
Checksum: 0,
},
},
{
User: "fake",
Fingerprint: model.Fingerprint(labels.StableHash(mustParseLabels(`{foo="bar"}`))),
Start: 1,
End: 4,
Checksum: 1,
ChunkRef: logproto.ChunkRef{
UserID: "fake",
Fingerprint: labels.StableHash(mustParseLabels(`{foo="bar"}`)),
From: 1,
Through: 4,
Checksum: 1,
},
},
{
User: "fake",
Fingerprint: model.Fingerprint(labels.StableHash(mustParseLabels(`{foo="bar"}`))),
Start: 2,
End: 5,
Checksum: 2,
ChunkRef: logproto.ChunkRef{
UserID: "fake",
Fingerprint: labels.StableHash(mustParseLabels(`{foo="bar"}`)),
From: 2,
Through: 5,
Checksum: 2,
},
},
{
User: "fake",
Fingerprint: model.Fingerprint(labels.StableHash(mustParseLabels(`{foo="bar", bazz="buzz"}`))),
Start: 1,
End: 10,
Checksum: 3,
ChunkRef: logproto.ChunkRef{
UserID: "fake",
Fingerprint: labels.StableHash(mustParseLabels(`{foo="bar", bazz="buzz"}`)),
From: 1,
Through: 10,
Checksum: 3,
},
},
}
require.Equal(t, expected, refs)
@@ -134,12 +142,14 @@ func TestSingleIdx(t *testing.T) {
require.Nil(t, err)
require.Equal(t, []ChunkRef{{
User: "fake",
Fingerprint: model.Fingerprint(labels.StableHash(mustParseLabels(`{foo="bar", bazz="buzz"}`))),
Start: 1,
End: 10,
Checksum: 3,
require.Equal(t, []logproto.ChunkRefWithSizingInfo{{
ChunkRef: logproto.ChunkRef{
UserID: "fake",
Fingerprint: labels.StableHash(mustParseLabels(`{foo="bar", bazz="buzz"}`)),
From: 1,
Through: 10,
Checksum: 3,
},
}}, shardedRefs)
})
pkg/storage/util_test.go
+8
View File
@@ -280,6 +280,14 @@ func (m *mockChunkStore) HasForSeries(_, _ model.Time) (sharding.ForSeries, bool
return nil, false
}
func (m *mockChunkStore) HasChunkSizingInfo(_, _ model.Time) bool {
return false
}
func (m *mockChunkStore) GetChunkRefsWithSizingInfo(_ context.Context, _ string, _, _ model.Time, _ chunk.Predicate) ([]logproto.ChunkRefWithSizingInfo, error) {
return nil, nil
}
func (m *mockChunkStore) Volume(_ context.Context, _ string, _, _ model.Time, _ int32, _ []string, _ string, _ ...*labels.Matcher) (*logproto.VolumeResponse, error) {
return nil, nil
}
tools/tsdb/index-analyzer/analytics.go
+2 -1
View File
@@ -7,6 +7,7 @@ import (
"github.com/prometheus/common/model"
"github.com/prometheus/prometheus/model/labels"
"github.com/grafana/loki/v3/pkg/logproto"
"github.com/grafana/loki/v3/pkg/storage/stores/shipper/indexshipper"
"github.com/grafana/loki/v3/pkg/storage/stores/shipper/indexshipper/index"
"github.com/grafana/loki/v3/pkg/storage/stores/shipper/indexshipper/tsdb"
@@ -19,7 +20,7 @@ func analyze(indexShipper indexshipper.IndexShipper, tableName string, tenants [
series int
chunks int
seriesRes []tsdb.Series
chunkRes []tsdb.ChunkRef
chunkRes []logproto.ChunkRefWithSizingInfo
maxChunksPerSeries int
seriesOver1kChunks int
)