Skip to content

Revert "Let forward requests run until timeout (#2679) + style fixes. #2771

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

Closed
wants to merge 2 commits into from
Closed

Revert "Let forward requests run until timeout (#2679) + style fixes. #2771

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4d35413
Revert "Let forward requests run until timeout (#2679)"
bwplotka Jun 15, 2020
4f822de
Minor style fixes.
bwplotka Jun 16, 2020
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
14 changes: 4 additions & 10 deletions pkg/receive/handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -230,7 +230,7 @@ func (h *Handler) handleRequest(ctx context.Context, rep uint64, tenant string,
replicated: rep != 0,
}

// on-the-wire format is 1-indexed and in-code is 0-indexed so we decrement the value if it was already replicated.
// On-the-wire format is 1-indexed and in-code is 0-indexed so we decrement the value if it was already replicated.
if r.replicated {
r.n--
}
Expand Down Expand Up @@ -450,7 +450,7 @@ func (h *Handler) fanoutForward(ctx context.Context, tenant string, replicas map
close(ec)
}()

// At the end, make sure to exhaust the channel, letting remaining unnecessary requests finish asnychronously.
// At the end, make sure to exhaust the channel, letting remaining unnecessary requests finish asynchronously.
// This is needed if context is cancelled or if we reached success of fail quorum faster.
defer func() {
go func() {
Expand Down Expand Up @@ -517,17 +517,11 @@ func (h *Handler) replicate(ctx context.Context, tenant string, wreq *prompb.Wri
}
h.mtx.RUnlock()

var err error
ctx, cancel := context.WithTimeout(ctx, h.options.ForwardTimeout)
defer func() {
// If there is no error, let forward requests optimistically run until timeout.
if err != nil {
cancel()
}
}()
defer cancel()

quorum := h.writeQuorum()
err = h.fanoutForward(ctx, tenant, replicas, wreqs, quorum)
err := h.fanoutForward(ctx, tenant, replicas, wreqs, quorum)
if countCause(err, isNotReady) >= quorum {
return tsdb.ErrNotReady
}
Expand Down