Skip to content

Commit 182165a

Browse files
salvacortssalvacorts
authored
chore: [k212] fix: Keep blocks referenced by newer metas (#13627)
1 parent c05a0e7 commit 182165a

File tree

2 files changed

+125
-16
lines changed

2 files changed

+125
-16
lines changed

‎pkg/bloombuild/planner/planner.go

+33-8
Original file line numberDiff line numberDiff line change
@@ -360,7 +360,7 @@ func (p *Planner) computeTasks(
360360

361361
// In case the planner restarted before deleting outdated metas in the previous iteration,
362362
// we delete them during the planning phase to avoid reprocessing them.
363-
metas, err = p.deleteOutdatedMetasAndBlocks(ctx, table, tenant, metas, phasePlanning)
363+
metas, err = p.deleteOutdatedMetasAndBlocks(ctx, table, tenant, nil, metas, phasePlanning)
364364
if err != nil {
365365
return nil, nil, fmt.Errorf("failed to delete outdated metas during planning: %w", err)
366366
}
@@ -446,8 +446,7 @@ func (p *Planner) processTenantTaskResults(
446446
return tasksSucceed, nil
447447
}
448448

449-
combined := append(originalMetas, newMetas...)
450-
if _, err := p.deleteOutdatedMetasAndBlocks(ctx, table, tenant, combined, phaseBuilding); err != nil {
449+
if _, err := p.deleteOutdatedMetasAndBlocks(ctx, table, tenant, newMetas, originalMetas, phaseBuilding); err != nil {
451450
return 0, fmt.Errorf("failed to delete outdated metas: %w", err)
452451
}
453452

@@ -460,12 +459,14 @@ func (p *Planner) deleteOutdatedMetasAndBlocks(
460459
ctx context.Context,
461460
table config.DayTable,
462461
tenant string,
463-
metas []bloomshipper.Meta,
462+
newMetas []bloomshipper.Meta,
463+
originalMetas []bloomshipper.Meta,
464464
phase string,
465465
) ([]bloomshipper.Meta, error) {
466466
logger := log.With(p.logger, "table", table.Addr(), "tenant", tenant, "phase", phase)
467467

468-
upToDate, outdated := outdatedMetas(metas)
468+
combined := append(originalMetas, newMetas...)
469+
upToDate, outdated := outdatedMetas(combined)
469470
if len(outdated) == 0 {
470471
level.Debug(logger).Log(
471472
"msg", "no outdated metas found",
@@ -497,17 +498,25 @@ func (p *Planner) deleteOutdatedMetasAndBlocks(
497498

498499
for _, meta := range outdated {
499500
for _, block := range meta.Blocks {
501+
logger := log.With(logger, "block", block.String())
502+
503+
// Prevent deleting blocks that are reused in new metas
504+
if isBlockInMetas(block, upToDate) {
505+
level.Debug(logger).Log("msg", "block is still in use in new meta, skipping delete")
506+
continue
507+
}
508+
500509
if err := client.DeleteBlocks(ctx, []bloomshipper.BlockRef{block}); err != nil {
501510
if client.IsObjectNotFoundErr(err) {
502-
level.Debug(logger).Log("msg", "block not found while attempting delete, continuing", "block", block.String())
511+
level.Debug(logger).Log("msg", "block not found while attempting delete, continuing")
503512
} else {
504-
level.Error(logger).Log("msg", "failed to delete block", "err", err, "block", block.String())
513+
level.Error(logger).Log("msg", "failed to delete block", "err", err)
505514
return nil, errors.Wrap(err, "failed to delete block")
506515
}
507516
}
508517

509518
deletedBlocks++
510-
level.Debug(logger).Log("msg", "removed outdated block", "block", block.String())
519+
level.Debug(logger).Log("msg", "removed outdated block")
511520
}
512521

513522
err = client.DeleteMetas(ctx, []bloomshipper.MetaRef{meta.MetaRef})
@@ -532,6 +541,22 @@ func (p *Planner) deleteOutdatedMetasAndBlocks(
532541
return upToDate, nil
533542
}
534543

544+
func isBlockInMetas(block bloomshipper.BlockRef, metas []bloomshipper.Meta) bool {
545+
// Blocks are sorted within a meta, so we can find it with binary search
546+
for _, meta := range metas {
547+
// Search for the first block whose bound is >= than the target block min bound.
548+
i := sort.Search(len(meta.Blocks), func(i int) bool {
549+
return meta.Blocks[i].Cmp(uint64(block.Bounds.Max)) <= v1.Overlap
550+
})
551+
552+
if i < len(meta.Blocks) && meta.Blocks[i] == block {
553+
return true
554+
}
555+
}
556+
557+
return false
558+
}
559+
535560
func (p *Planner) tables(ts time.Time) *dayRangeIterator {
536561
// adjust the minimum by one to make it inclusive, which is more intuitive
537562
// for a configuration variable

‎pkg/bloombuild/planner/planner_test.go

+92-8
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package planner
22

33
import (
4+
"bytes"
45
"context"
56
"fmt"
67
"io"
@@ -20,6 +21,8 @@ import (
2021
"google.golang.org/grpc"
2122

2223
"github.com/grafana/loki/v3/pkg/bloombuild/protos"
24+
"github.com/grafana/loki/v3/pkg/chunkenc"
25+
iter "github.com/grafana/loki/v3/pkg/iter/v2"
2326
"github.com/grafana/loki/v3/pkg/storage"
2427
v1 "github.com/grafana/loki/v3/pkg/storage/bloom/v1"
2528
"github.com/grafana/loki/v3/pkg/storage/chunk/cache"
@@ -166,11 +169,36 @@ func genBlockRef(min, max model.Fingerprint) bloomshipper.BlockRef {
166169
}
167170
}
168171

169-
func genBlock(ref bloomshipper.BlockRef) bloomshipper.Block {
172+
func genBlock(ref bloomshipper.BlockRef) (bloomshipper.Block, error) {
173+
indexBuf := bytes.NewBuffer(nil)
174+
bloomsBuf := bytes.NewBuffer(nil)
175+
writer := v1.NewMemoryBlockWriter(indexBuf, bloomsBuf)
176+
reader := v1.NewByteReader(indexBuf, bloomsBuf)
177+
178+
blockOpts := v1.NewBlockOptions(chunkenc.EncNone, 4, 1, 0, 0)
179+
180+
builder, err := v1.NewBlockBuilder(blockOpts, writer)
181+
if err != nil {
182+
return bloomshipper.Block{}, err
183+
}
184+
185+
if _, err = builder.BuildFrom(iter.NewEmptyIter[v1.SeriesWithBlooms]()); err != nil {
186+
return bloomshipper.Block{}, err
187+
}
188+
189+
block := v1.NewBlock(reader, v1.NewMetrics(nil))
190+
191+
buf := bytes.NewBuffer(nil)
192+
if err := v1.TarGz(buf, block.Reader()); err != nil {
193+
return bloomshipper.Block{}, err
194+
}
195+
196+
tarReader := bytes.NewReader(buf.Bytes())
197+
170198
return bloomshipper.Block{
171199
BlockRef: ref,
172-
Data: &DummyReadSeekCloser{},
173-
}
200+
Data: bloomshipper.ClosableReadSeekerAdapter{ReadSeeker: tarReader},
201+
}, nil
174202
}
175203

176204
func Test_blockPlansForGaps(t *testing.T) {
@@ -612,7 +640,12 @@ func putMetas(bloomClient bloomshipper.Client, metas []bloomshipper.Meta) error
612640
}
613641

614642
for _, block := range meta.Blocks {
615-
err := bloomClient.PutBlock(context.Background(), genBlock(block))
643+
writtenBlock, err := genBlock(block)
644+
if err != nil {
645+
return err
646+
}
647+
648+
err = bloomClient.PutBlock(context.Background(), writtenBlock)
616649
if err != nil {
617650
return err
618651
}
@@ -826,6 +859,7 @@ func Test_deleteOutdatedMetas(t *testing.T) {
826859
for _, tc := range []struct {
827860
name string
828861
originalMetas []bloomshipper.Meta
862+
newMetas []bloomshipper.Meta
829863
expectedUpToDateMetas []bloomshipper.Meta
830864
}{
831865
{
@@ -835,6 +869,8 @@ func Test_deleteOutdatedMetas(t *testing.T) {
835869
name: "only up to date metas",
836870
originalMetas: []bloomshipper.Meta{
837871
genMeta(0, 10, []int{0}, []bloomshipper.BlockRef{genBlockRef(0, 10)}),
872+
},
873+
newMetas: []bloomshipper.Meta{
838874
genMeta(10, 20, []int{0}, []bloomshipper.BlockRef{genBlockRef(10, 20)}),
839875
},
840876
expectedUpToDateMetas: []bloomshipper.Meta{
@@ -846,13 +882,52 @@ func Test_deleteOutdatedMetas(t *testing.T) {
846882
name: "outdated metas",
847883
originalMetas: []bloomshipper.Meta{
848884
genMeta(0, 5, []int{0}, []bloomshipper.BlockRef{genBlockRef(0, 5)}),
849-
genMeta(6, 10, []int{0}, []bloomshipper.BlockRef{genBlockRef(6, 10)}),
885+
},
886+
newMetas: []bloomshipper.Meta{
850887
genMeta(0, 10, []int{1}, []bloomshipper.BlockRef{genBlockRef(0, 10)}),
851888
},
852889
expectedUpToDateMetas: []bloomshipper.Meta{
853890
genMeta(0, 10, []int{1}, []bloomshipper.BlockRef{genBlockRef(0, 10)}),
854891
},
855892
},
893+
{
894+
name: "new metas reuse blocks from outdated meta",
895+
originalMetas: []bloomshipper.Meta{
896+
genMeta(0, 10, []int{0}, []bloomshipper.BlockRef{ // Outdated
897+
genBlockRef(0, 5), // Reuse
898+
genBlockRef(5, 10), // Delete
899+
}),
900+
genMeta(10, 20, []int{0}, []bloomshipper.BlockRef{ // Outdated
901+
genBlockRef(10, 20), // Reuse
902+
}),
903+
genMeta(20, 30, []int{0}, []bloomshipper.BlockRef{ // Up to date
904+
genBlockRef(20, 30),
905+
}),
906+
},
907+
newMetas: []bloomshipper.Meta{
908+
genMeta(0, 5, []int{1}, []bloomshipper.BlockRef{
909+
genBlockRef(0, 5), // Reused block
910+
}),
911+
genMeta(5, 20, []int{1}, []bloomshipper.BlockRef{
912+
genBlockRef(5, 7), // New block
913+
genBlockRef(7, 10), // New block
914+
genBlockRef(10, 20), // Reused block
915+
}),
916+
},
917+
expectedUpToDateMetas: []bloomshipper.Meta{
918+
genMeta(0, 5, []int{1}, []bloomshipper.BlockRef{
919+
genBlockRef(0, 5),
920+
}),
921+
genMeta(5, 20, []int{1}, []bloomshipper.BlockRef{
922+
genBlockRef(5, 7),
923+
genBlockRef(7, 10),
924+
genBlockRef(10, 20),
925+
}),
926+
genMeta(20, 30, []int{0}, []bloomshipper.BlockRef{
927+
genBlockRef(20, 30),
928+
}),
929+
},
930+
},
856931
} {
857932
t.Run(tc.name, func(t *testing.T) {
858933
logger := log.NewNopLogger()
@@ -867,9 +942,11 @@ func Test_deleteOutdatedMetas(t *testing.T) {
867942
bloomClient, err := planner.bloomStore.Client(testDay.ModelTime())
868943
require.NoError(t, err)
869944

870-
// Create original metas and blocks
945+
// Create original/new metas and blocks
871946
err = putMetas(bloomClient, tc.originalMetas)
872947
require.NoError(t, err)
948+
err = putMetas(bloomClient, tc.newMetas)
949+
require.NoError(t, err)
873950

874951
// Get all metas
875952
metas, err := planner.bloomStore.FetchMetas(
@@ -882,9 +959,9 @@ func Test_deleteOutdatedMetas(t *testing.T) {
882959
)
883960
require.NoError(t, err)
884961
removeLocFromMetasSources(metas)
885-
require.ElementsMatch(t, tc.originalMetas, metas)
962+
require.ElementsMatch(t, append(tc.originalMetas, tc.newMetas...), metas)
886963

887-
upToDate, err := planner.deleteOutdatedMetasAndBlocks(context.Background(), testTable, "fakeTenant", tc.originalMetas, phasePlanning)
964+
upToDate, err := planner.deleteOutdatedMetasAndBlocks(context.Background(), testTable, "fakeTenant", tc.newMetas, tc.originalMetas, phasePlanning)
888965
require.NoError(t, err)
889966
require.ElementsMatch(t, tc.expectedUpToDateMetas, upToDate)
890967

@@ -900,6 +977,13 @@ func Test_deleteOutdatedMetas(t *testing.T) {
900977
require.NoError(t, err)
901978
removeLocFromMetasSources(metas)
902979
require.ElementsMatch(t, tc.expectedUpToDateMetas, metas)
980+
981+
// Fetch all blocks from the metas
982+
for _, meta := range metas {
983+
blocks, err := planner.bloomStore.FetchBlocks(context.Background(), meta.Blocks)
984+
require.NoError(t, err)
985+
require.Len(t, blocks, len(meta.Blocks))
986+
}
903987
})
904988
}
905989
}

0 commit comments

Comments
 (0)