mirror of https://github.com/grafana/loki
refactor(blooms): Delete outdated metas (#13153)
Salva Corts
12 months ago
committed by
GitHub
parent
fbe7c559b5
commit
9c96d26895
@ -0,0 +1,247 @@ |
||||
package planner |
||||
|
||||
import ( |
||||
"sort" |
||||
|
||||
"github.com/prometheus/common/model" |
||||
|
||||
v1 "github.com/grafana/loki/v3/pkg/storage/bloom/v1" |
||||
"github.com/grafana/loki/v3/pkg/storage/stores/shipper/bloomshipper" |
||||
) |
||||
|
||||
type tsdbToken struct { |
||||
through model.Fingerprint // inclusive
|
||||
version int // TSDB version
|
||||
} |
||||
|
||||
// a ring of token ranges used to identify old metas.
|
||||
// each token represents that a TSDB version has covered the entire range
|
||||
// up to that point from the previous token.
|
||||
type tsdbTokenRange []tsdbToken |
||||
|
||||
func (t tsdbTokenRange) Len() int { |
||||
return len(t) |
||||
} |
||||
|
||||
func (t tsdbTokenRange) Less(i, j int) bool { |
||||
return t[i].through < t[j].through |
||||
} |
||||
|
||||
func (t tsdbTokenRange) Swap(i, j int) { |
||||
t[i], t[j] = t[j], t[i] |
||||
} |
||||
|
||||
// Add ensures a versioned set of bounds is added to the range. If the bounds are already
|
||||
// covered by a more up to date version, it returns false.
|
||||
func (t tsdbTokenRange) Add(version int, bounds v1.FingerprintBounds) (res tsdbTokenRange, added bool) { |
||||
// allows attempting to join neighboring token ranges with identical versions
|
||||
// that aren't known until the end of the function
|
||||
var shouldReassemble bool |
||||
var reassembleFrom int |
||||
defer func() { |
||||
if shouldReassemble { |
||||
res = res.reassemble(reassembleFrom) |
||||
} |
||||
}() |
||||
|
||||
// special case: first token
|
||||
if len(t) == 0 { |
||||
tok := tsdbToken{through: bounds.Max, version: version} |
||||
// special case: first token is included in bounds, no need to fill negative space
|
||||
if bounds.Min == 0 { |
||||
return append(t, tok), true |
||||
} |
||||
// Use a negative version to indicate that the range is not covered by any version.
|
||||
return append(t, tsdbToken{through: bounds.Min - 1, version: -1}, tok), true |
||||
} |
||||
|
||||
// For non-nil token ranges, we continually update the range with newer versions.
|
||||
for { |
||||
// find first token that covers the start of the range
|
||||
i := sort.Search(len(t), func(i int) bool { |
||||
return t[i].through >= bounds.Min |
||||
}) |
||||
|
||||
if i == len(t) { |
||||
tok := tsdbToken{through: bounds.Max, version: version} |
||||
|
||||
// edge case: there is no gap between the previous token range
|
||||
// and the new one;
|
||||
// skip adding a negative token
|
||||
if t[len(t)-1].through == bounds.Min-1 { |
||||
return append(t, tok), true |
||||
} |
||||
|
||||
// the range is not covered by any version and we are at the end of the range.
|
||||
// Add a negative token and the new token.
|
||||
negative := tsdbToken{through: bounds.Min - 1, version: -1} |
||||
return append(t, negative, tok), true |
||||
} |
||||
|
||||
// Otherwise, we've found a token that covers the start of the range.
|
||||
newer := t[i].version < version |
||||
preExisting := t.boundsForToken(i) |
||||
if !newer { |
||||
if bounds.Within(preExisting) { |
||||
// The range is already covered by a more up to date version, no need
|
||||
// to add anything, but honor if an earlier token was added
|
||||
return t, added |
||||
} |
||||
|
||||
// The range is partially covered by a more up to date version;
|
||||
// update the range we need to check and continue
|
||||
bounds = v1.NewBounds(preExisting.Max+1, bounds.Max) |
||||
continue |
||||
} |
||||
|
||||
// If we need to update the range, there are 5 cases:
|
||||
// 1. `equal`: the incoming range equals an existing range ()
|
||||
// ------ # addition
|
||||
// ------ # src
|
||||
// 2. `subset`: the incoming range is a subset of an existing range
|
||||
// ------ # addition
|
||||
// -------- # src
|
||||
// 3. `overflow_both_sides`: the incoming range is a superset of an existing range. This is not possible
|
||||
// because the first token in the ring implicitly covers the left bound (zero) of all possible fps.
|
||||
// Therefore, we can skip this case.
|
||||
// ------ # addition
|
||||
// ---- # src
|
||||
// 4. `right_overflow`: the incoming range overflows the right side of an existing range
|
||||
// ------ # addition
|
||||
// ------ # src
|
||||
// 5. `left_overflow`: the incoming range overflows the left side of an existing range. This can be skipped
|
||||
// for the same reason as `superset`.
|
||||
// ------ # addition
|
||||
// ------ # src
|
||||
|
||||
// 1) (`equal`): we're replacing the same bounds
|
||||
if bounds.Equal(preExisting) { |
||||
t[i].version = version |
||||
return t, true |
||||
} |
||||
|
||||
// 2) (`subset`): the incoming range is a subset of an existing range
|
||||
if bounds.Within(preExisting) { |
||||
// 2a) the incoming range touches the existing range's minimum bound
|
||||
if bounds.Min == preExisting.Min { |
||||
tok := tsdbToken{through: bounds.Max, version: version} |
||||
t = append(t, tsdbToken{}) |
||||
copy(t[i+1:], t[i:]) |
||||
t[i] = tok |
||||
return t, true |
||||
} |
||||
// 2b) the incoming range touches the existing range's maximum bound
|
||||
if bounds.Max == preExisting.Max { |
||||
t[i].through = bounds.Min - 1 |
||||
tok := tsdbToken{through: bounds.Max, version: version} |
||||
t = append(t, tsdbToken{}) |
||||
copy(t[i+2:], t[i+1:]) |
||||
t[i+1] = tok |
||||
return t, true |
||||
} |
||||
|
||||
// 2c) the incoming range is does not touch either edge;
|
||||
// add two tokens (the new one and a new left-bound for the old range)
|
||||
tok := tsdbToken{through: bounds.Max, version: version} |
||||
t = append(t, tsdbToken{}, tsdbToken{}) |
||||
copy(t[i+2:], t[i:]) |
||||
t[i+1] = tok |
||||
t[i].through = bounds.Min - 1 |
||||
return t, true |
||||
} |
||||
|
||||
// 4) (`right_overflow`): the incoming range overflows the right side of an existing range
|
||||
|
||||
// 4a) shortcut: the incoming range is a right-overlapping superset of the existing range.
|
||||
// replace the existing token's version, update reassembly targets for merging neighboring ranges
|
||||
// w/ the same version, and continue
|
||||
if preExisting.Min == bounds.Min { |
||||
t[i].version = version |
||||
bounds.Min = preExisting.Max + 1 |
||||
added = true |
||||
if !shouldReassemble { |
||||
reassembleFrom = i |
||||
shouldReassemble = true |
||||
} |
||||
continue |
||||
} |
||||
|
||||
// 4b) the incoming range overlaps the right side of the existing range but
|
||||
// does not touch the left side;
|
||||
// add a new token for the right side of the existing range then update the reassembly targets
|
||||
// and continue
|
||||
overlap := tsdbToken{through: t[i].through, version: version} |
||||
t[i].through = bounds.Min - 1 |
||||
t = append(t, tsdbToken{}) |
||||
copy(t[i+2:], t[i+1:]) |
||||
t[i+1] = overlap |
||||
added = true |
||||
bounds.Min = overlap.through + 1 |
||||
if !shouldReassemble { |
||||
reassembleFrom = i + 1 |
||||
shouldReassemble = true |
||||
} |
||||
continue |
||||
} |
||||
} |
||||
|
||||
func (t tsdbTokenRange) boundsForToken(i int) v1.FingerprintBounds { |
||||
if i == 0 { |
||||
return v1.FingerprintBounds{Min: 0, Max: t[i].through} |
||||
} |
||||
return v1.FingerprintBounds{Min: t[i-1].through + 1, Max: t[i].through} |
||||
} |
||||
|
||||
// reassemble merges neighboring tokens with the same version
|
||||
func (t tsdbTokenRange) reassemble(from int) tsdbTokenRange { |
||||
reassembleTo := from |
||||
for i := from; i < len(t)-1; i++ { |
||||
if t[i].version != t[i+1].version { |
||||
break |
||||
} |
||||
reassembleTo = i + 1 |
||||
} |
||||
|
||||
if reassembleTo == from { |
||||
return t |
||||
} |
||||
t[from].through = t[reassembleTo].through |
||||
copy(t[from+1:], t[reassembleTo+1:]) |
||||
return t[:len(t)-(reassembleTo-from)] |
||||
} |
||||
|
||||
func outdatedMetas(metas []bloomshipper.Meta) (outdated []bloomshipper.Meta, err error) { |
||||
// Sort metas descending by most recent source when checking
|
||||
// for outdated metas (older metas are discarded if they don't change the range).
|
||||
sort.Slice(metas, func(i, j int) bool { |
||||
a, err := metas[i].MostRecentSource() |
||||
if err != nil { |
||||
panic(err.Error()) |
||||
} |
||||
b, err := metas[j].MostRecentSource() |
||||
if err != nil { |
||||
panic(err.Error()) |
||||
} |
||||
return !a.TS.Before(b.TS) |
||||
}) |
||||
|
||||
var ( |
||||
tokenRange tsdbTokenRange |
||||
added bool |
||||
) |
||||
|
||||
for _, meta := range metas { |
||||
mostRecent, err := meta.MostRecentSource() |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
version := int(model.TimeFromUnixNano(mostRecent.TS.UnixNano())) |
||||
tokenRange, added = tokenRange.Add(version, meta.Bounds) |
||||
if !added { |
||||
outdated = append(outdated, meta) |
||||
} |
||||
} |
||||
|
||||
return outdated, nil |
||||
|
||||
} |
||||
@ -0,0 +1,323 @@ |
||||
package planner |
||||
|
||||
import ( |
||||
"testing" |
||||
|
||||
"github.com/prometheus/common/model" |
||||
"github.com/stretchr/testify/require" |
||||
|
||||
v1 "github.com/grafana/loki/v3/pkg/storage/bloom/v1" |
||||
"github.com/grafana/loki/v3/pkg/storage/stores/shipper/bloomshipper" |
||||
"github.com/grafana/loki/v3/pkg/storage/stores/shipper/indexshipper/tsdb" |
||||
) |
||||
|
||||
func Test_TsdbTokenRange(t *testing.T) { |
||||
type addition struct { |
||||
version int |
||||
bounds v1.FingerprintBounds |
||||
} |
||||
type exp struct { |
||||
added bool |
||||
err bool |
||||
} |
||||
mk := func(version int, min, max model.Fingerprint) addition { |
||||
return addition{version, v1.FingerprintBounds{Min: min, Max: max}} |
||||
} |
||||
tok := func(version int, through model.Fingerprint) tsdbToken { |
||||
return tsdbToken{version: version, through: through} |
||||
} |
||||
|
||||
for _, tc := range []struct { |
||||
desc string |
||||
additions []addition |
||||
exp []bool |
||||
result tsdbTokenRange |
||||
}{ |
||||
{ |
||||
desc: "ascending versions", |
||||
additions: []addition{ |
||||
mk(1, 0, 10), |
||||
mk(2, 11, 20), |
||||
mk(3, 15, 25), |
||||
}, |
||||
exp: []bool{true, true, true}, |
||||
result: tsdbTokenRange{ |
||||
tok(1, 10), |
||||
tok(2, 14), |
||||
tok(3, 25), |
||||
}, |
||||
}, |
||||
{ |
||||
desc: "descending versions", |
||||
additions: []addition{ |
||||
mk(3, 15, 25), |
||||
mk(2, 11, 20), |
||||
mk(1, 0, 10), |
||||
}, |
||||
exp: []bool{true, true, true}, |
||||
result: tsdbTokenRange{ |
||||
tok(1, 10), |
||||
tok(2, 14), |
||||
tok(3, 25), |
||||
}, |
||||
}, |
||||
{ |
||||
desc: "simple", |
||||
additions: []addition{ |
||||
mk(3, 0, 10), |
||||
mk(2, 11, 20), |
||||
mk(1, 15, 25), |
||||
}, |
||||
exp: []bool{true, true, true}, |
||||
result: tsdbTokenRange{ |
||||
tok(3, 10), |
||||
tok(2, 20), |
||||
tok(1, 25), |
||||
}, |
||||
}, |
||||
{ |
||||
desc: "simple replacement", |
||||
additions: []addition{ |
||||
mk(3, 10, 20), |
||||
mk(2, 0, 9), |
||||
}, |
||||
exp: []bool{true, true}, |
||||
result: tsdbTokenRange{ |
||||
tok(2, 9), |
||||
tok(3, 20), |
||||
}, |
||||
}, |
||||
{ |
||||
desc: "complex", |
||||
additions: []addition{ |
||||
mk(5, 30, 50), |
||||
mk(4, 20, 45), |
||||
mk(3, 25, 70), |
||||
mk(2, 10, 20), |
||||
mk(1, 1, 5), |
||||
}, |
||||
exp: []bool{true, true, true, true, true, true}, |
||||
result: tsdbTokenRange{ |
||||
tok(-1, 0), |
||||
tok(1, 5), |
||||
tok(-1, 9), |
||||
tok(2, 19), |
||||
tok(4, 29), |
||||
tok(5, 50), |
||||
tok(3, 70), |
||||
}, |
||||
}, |
||||
{ |
||||
desc: "neighboring upper range", |
||||
additions: []addition{ |
||||
mk(5, 30, 50), |
||||
mk(4, 51, 60), |
||||
}, |
||||
exp: []bool{true, true}, |
||||
result: tsdbTokenRange{ |
||||
tok(-1, 29), |
||||
tok(5, 50), |
||||
tok(4, 60), |
||||
}, |
||||
}, |
||||
{ |
||||
desc: "non-neighboring upper range", |
||||
additions: []addition{ |
||||
mk(5, 30, 50), |
||||
mk(4, 55, 60), |
||||
}, |
||||
exp: []bool{true, true}, |
||||
result: tsdbTokenRange{ |
||||
tok(-1, 29), |
||||
tok(5, 50), |
||||
tok(-1, 54), |
||||
tok(4, 60), |
||||
}, |
||||
}, |
||||
{ |
||||
desc: "earlier version within", |
||||
additions: []addition{ |
||||
mk(5, 30, 50), |
||||
mk(4, 40, 45), |
||||
}, |
||||
exp: []bool{true, false}, |
||||
result: tsdbTokenRange{ |
||||
tok(-1, 29), |
||||
tok(5, 50), |
||||
}, |
||||
}, |
||||
{ |
||||
desc: "earlier version right overlapping", |
||||
additions: []addition{ |
||||
mk(5, 10, 20), |
||||
mk(4, 15, 25), |
||||
}, |
||||
exp: []bool{true, true}, |
||||
result: tsdbTokenRange{ |
||||
tok(-1, 9), |
||||
tok(5, 20), |
||||
tok(4, 25), |
||||
}, |
||||
}, |
||||
{ |
||||
desc: "older version overlaps two", |
||||
additions: []addition{ |
||||
mk(3, 10, 20), |
||||
mk(2, 21, 30), |
||||
mk(1, 15, 25), |
||||
}, |
||||
exp: []bool{true, true, false}, |
||||
result: tsdbTokenRange{ |
||||
tok(-1, 9), |
||||
tok(3, 20), |
||||
tok(2, 30), |
||||
}, |
||||
}, |
||||
{ |
||||
desc: "older version overlaps two w middle", |
||||
additions: []addition{ |
||||
mk(3, 10, 20), |
||||
mk(2, 22, 30), |
||||
mk(1, 15, 25), |
||||
}, |
||||
exp: []bool{true, true, true}, |
||||
result: tsdbTokenRange{ |
||||
tok(-1, 9), |
||||
tok(3, 20), |
||||
tok(1, 21), |
||||
tok(2, 30), |
||||
}, |
||||
}, |
||||
{ |
||||
desc: "newer right overflow", |
||||
additions: []addition{ |
||||
mk(1, 30, 50), |
||||
mk(2, 40, 60), |
||||
}, |
||||
exp: []bool{true, true}, |
||||
result: tsdbTokenRange{ |
||||
tok(-1, 29), |
||||
tok(1, 39), |
||||
tok(2, 60), |
||||
}, |
||||
}, |
||||
{ |
||||
desc: "newer right overflow superset", |
||||
additions: []addition{ |
||||
mk(1, 30, 50), |
||||
mk(2, 30, 60), |
||||
}, |
||||
exp: []bool{true, true}, |
||||
result: tsdbTokenRange{ |
||||
tok(-1, 29), |
||||
tok(2, 60), |
||||
}, |
||||
}, |
||||
{ |
||||
desc: "newer right overflow partial", |
||||
additions: []addition{ |
||||
mk(1, 30, 50), |
||||
mk(2, 40, 60), |
||||
}, |
||||
exp: []bool{true, true}, |
||||
result: tsdbTokenRange{ |
||||
tok(-1, 29), |
||||
tok(1, 39), |
||||
tok(2, 60), |
||||
}, |
||||
}, |
||||
} { |
||||
t.Run(tc.desc, func(t *testing.T) { |
||||
var ( |
||||
tr tsdbTokenRange |
||||
added bool |
||||
) |
||||
for i, a := range tc.additions { |
||||
tr, added = tr.Add(a.version, a.bounds) |
||||
exp := tc.exp[i] |
||||
require.Equal(t, exp, added, "on iteration %d", i) |
||||
} |
||||
require.Equal(t, tc.result, tr) |
||||
}) |
||||
} |
||||
} |
||||
|
||||
func Test_OutdatedMetas(t *testing.T) { |
||||
gen := func(bounds v1.FingerprintBounds, tsdbTimes ...model.Time) (meta bloomshipper.Meta) { |
||||
for _, tsdbTime := range tsdbTimes { |
||||
meta.Sources = append(meta.Sources, tsdb.SingleTenantTSDBIdentifier{TS: tsdbTime.Time()}) |
||||
} |
||||
meta.Bounds = bounds |
||||
return meta |
||||
} |
||||
|
||||
for _, tc := range []struct { |
||||
desc string |
||||
metas []bloomshipper.Meta |
||||
exp []bloomshipper.Meta |
||||
}{ |
||||
{ |
||||
desc: "no metas", |
||||
metas: nil, |
||||
exp: nil, |
||||
}, |
||||
{ |
||||
desc: "single meta", |
||||
metas: []bloomshipper.Meta{ |
||||
gen(v1.NewBounds(0, 10), 0), |
||||
}, |
||||
exp: nil, |
||||
}, |
||||
{ |
||||
desc: "single outdated meta", |
||||
metas: []bloomshipper.Meta{ |
||||
gen(v1.NewBounds(0, 10), 0), |
||||
gen(v1.NewBounds(0, 10), 1), |
||||
}, |
||||
exp: []bloomshipper.Meta{ |
||||
gen(v1.NewBounds(0, 10), 0), |
||||
}, |
||||
}, |
||||
{ |
||||
desc: "single outdated via partitions", |
||||
metas: []bloomshipper.Meta{ |
||||
gen(v1.NewBounds(0, 5), 0), |
||||
gen(v1.NewBounds(6, 10), 0), |
||||
gen(v1.NewBounds(0, 10), 1), |
||||
}, |
||||
exp: []bloomshipper.Meta{ |
||||
gen(v1.NewBounds(6, 10), 0), |
||||
gen(v1.NewBounds(0, 5), 0), |
||||
}, |
||||
}, |
||||
{ |
||||
desc: "same tsdb versions", |
||||
metas: []bloomshipper.Meta{ |
||||
gen(v1.NewBounds(0, 5), 0), |
||||
gen(v1.NewBounds(6, 10), 0), |
||||
gen(v1.NewBounds(0, 10), 1), |
||||
}, |
||||
exp: []bloomshipper.Meta{ |
||||
gen(v1.NewBounds(6, 10), 0), |
||||
gen(v1.NewBounds(0, 5), 0), |
||||
}, |
||||
}, |
||||
{ |
||||
desc: "multi version ordering", |
||||
metas: []bloomshipper.Meta{ |
||||
gen(v1.NewBounds(0, 5), 0), |
||||
gen(v1.NewBounds(0, 10), 1), // only part of the range is outdated, must keep
|
||||
gen(v1.NewBounds(8, 10), 2), |
||||
}, |
||||
exp: []bloomshipper.Meta{ |
||||
gen(v1.NewBounds(0, 5), 0), |
||||
}, |
||||
}, |
||||
} { |
||||
t.Run(tc.desc, func(t *testing.T) { |
||||
outdated, err := outdatedMetas(tc.metas) |
||||
require.NoError(t, err) |
||||
require.Equal(t, tc.exp, outdated) |
||||
}) |
||||
} |
||||
} |
||||
Loading…
Reference in new issue