Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(sharding): apply offset to both from and through in shard request #14256

Merged
merged 3 commits into from
Sep 27, 2024
Merged

fix(sharding): apply offset to both from and through in shard request #14256

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f1a8380
fix(sharding): offset both from and through in shard request
ashwanthgoli Sep 25, 2024
f91dfa2
fixup! fix(sharding): offset both from and through in shard request
ashwanthgoli Sep 25, 2024
72a65d4
fixup! fix(sharding): offset both from and through in shard request
ashwanthgoli Sep 25, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 20 additions & 5 deletions pkg/querier/queryrange/shard_resolver.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,7 +225,10 @@ func (r *dynamicShardResolver) ShardingRanges(expr syntax.Expr, targetBytesPerSh
) {
log := spanlogger.FromContext(r.ctx)

adjustedFrom := r.from
var (
adjustedFrom = r.from
adjustedThrough model.Time
)

// NB(owen-d): there should only ever be 1 matcher group passed
// to this call as we call it separately for different legs
Expand All @@ -236,18 +239,30 @@ func (r *dynamicShardResolver) ShardingRanges(expr syntax.Expr, targetBytesPerSh
}

for _, grp := range grps {
diff := grp.Interval + grp.Offset
diff := grp.Interval

// For instant queries, when start == end,
// we have a default lookback which we add here
if grp.Interval == 0 {
diff = diff + r.defaultLookback
if diff == 0 {
diff = r.defaultLookback
}

diff += grp.Offset

// use the oldest adjustedFrom
if r.from.Add(-diff).Before(adjustedFrom) {
adjustedFrom = r.from.Add(-diff)
}

// use the latest adjustedThrough
if r.through.Add(-grp.Offset).After(adjustedThrough) {
adjustedThrough = r.through.Add(-grp.Offset)
}
}

// handle the case where there are no matchers
if adjustedThrough == 0 {
adjustedThrough = r.through
}

exprStr := expr.String()
Expand All @@ -256,7 +271,7 @@ func (r *dynamicShardResolver) ShardingRanges(expr syntax.Expr, targetBytesPerSh
// use the retry handler here to retry transient errors
resp, err := r.retryNextHandler.Do(r.ctx, &logproto.ShardsRequest{
From: adjustedFrom,
Through: r.through,
Through: adjustedThrough,
Query: expr.String(),
TargetBytesPerShard: targetBytesPerShard,
})
Expand Down
Loading