-
Notifications
You must be signed in to change notification settings - Fork 2.1k
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.
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
query: add store.read-timeout parameter to avoid partial response failure when one of stores timed out #895
Changes from 15 commits
dfadec5
3a5701e
fc4460e
81d7d3d
5a36f3f
5906c73
0e8adab
0bdcf03
86a44da
07b419b
8356b80
0224bf3
7f75399
8ffee1b
9be840e
4de0dc0
afe009b
e5f2494
a1ef762
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -7,7 +7,6 @@ import ( | |
"math" | ||
"math/rand" | ||
"testing" | ||
|
||
"time" | ||
|
||
"github.com/fortytw2/leaktest" | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -33,6 +33,7 @@ type Client interface { | |
TimeRange() (mint int64, maxt int64) | ||
|
||
String() string | ||
|
||
// Addr returns address of a Client. | ||
Addr() string | ||
} | ||
|
@@ -44,7 +45,11 @@ type ProxyStore struct { | |
component component.StoreAPI | ||
selectorLabels labels.Labels | ||
|
||
// responseTimeout is a timeout for any GRPC operation during series query | ||
responseTimeout time.Duration | ||
|
||
// readTimeout is a timeout for entire store request | ||
readTimeout time.Duration | ||
} | ||
|
||
// NewProxyStore returns a new ProxyStore that uses the given clients that implements storeAPI to fan-in all series to the client. | ||
|
@@ -55,6 +60,7 @@ func NewProxyStore( | |
component component.StoreAPI, | ||
selectorLabels labels.Labels, | ||
responseTimeout time.Duration, | ||
readTimeout time.Duration, | ||
) *ProxyStore { | ||
if logger == nil { | ||
logger = log.NewNopLogger() | ||
|
@@ -66,6 +72,7 @@ func NewProxyStore( | |
component: component, | ||
selectorLabels: selectorLabels, | ||
responseTimeout: responseTimeout, | ||
readTimeout: readTimeout, | ||
} | ||
return s | ||
} | ||
|
@@ -119,6 +126,8 @@ func newRespCh(ctx context.Context, buffer int) (*ctxRespSender, <-chan *storepb | |
return &ctxRespSender{ctx: ctx, ch: respCh}, respCh, func() { close(respCh) } | ||
} | ||
|
||
// send writes response to sender channel | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Please use full sentence comments. They are for humans (: |
||
// or just returns if sender context was timed out | ||
func (s ctxRespSender) send(r *storepb.SeriesResponse) { | ||
select { | ||
case <-s.ctx.Done(): | ||
|
@@ -147,6 +156,9 @@ func (s *ProxyStore) Series(r *storepb.SeriesRequest, srv storepb.Store_SeriesSe | |
respSender, respRecv, closeFn = newRespCh(gctx, 10) | ||
) | ||
|
||
storeReadCtx, storeReadCancelFunc := s.contextWithReadTimeout(gctx) | ||
defer storeReadCancelFunc() | ||
|
||
g.Go(func() error { | ||
var ( | ||
seriesSet []storepb.SeriesSet | ||
|
@@ -178,7 +190,7 @@ func (s *ProxyStore) Series(r *storepb.SeriesRequest, srv storepb.Store_SeriesSe | |
storeDebugMsgs = append(storeDebugMsgs, fmt.Sprintf("store %s queried", st)) | ||
|
||
// This is used to cancel this stream when one operations takes too long. | ||
seriesCtx, closeSeries := context.WithCancel(gctx) | ||
seriesCtx, closeSeries := context.WithCancel(storeReadCtx) | ||
defer closeSeries() | ||
|
||
sc, err := st.Series(seriesCtx, r) | ||
|
@@ -295,19 +307,12 @@ func startStreamSeriesSet( | |
} | ||
|
||
if ctx.Err() != nil { | ||
s.writeWarningOrErrorResponse(partialResponse, errors.Wrapf(ctx.Err(), "receive series from %s", s.name)) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is nice 👍 |
||
return | ||
} | ||
|
||
if err != nil { | ||
wrapErr := errors.Wrapf(err, "receive series from %s", s.name) | ||
if partialResponse { | ||
s.warnCh.send(storepb.NewWarnSeriesResponse(wrapErr)) | ||
return | ||
} | ||
|
||
s.errMtx.Lock() | ||
s.err = wrapErr | ||
s.errMtx.Unlock() | ||
s.writeWarningOrErrorResponse(partialResponse, errors.Wrapf(err, "receive series from %s", s.name)) | ||
return | ||
} | ||
|
||
|
@@ -356,6 +361,24 @@ func (s *streamSeriesSet) Next() (ok bool) { | |
} | ||
} | ||
|
||
// writeWarningOrErrorResponse sends warning if partial response enabled or sets error otherwise | ||
func (s *streamSeriesSet) writeWarningOrErrorResponse(partialResponse bool, err error) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We can make this return nothing since we aren't using the return values anywhere and we aren't even setting them 😄 |
||
if partialResponse { | ||
s.warnCh.send(storepb.NewWarnSeriesResponse(err)) | ||
return | ||
} | ||
|
||
s.setError(err) | ||
return | ||
} | ||
|
||
// setError sets error (thread-safe) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. full sentence comments |
||
func (s *streamSeriesSet) setError(err error) { | ||
s.errMtx.Lock() | ||
s.err = err | ||
s.errMtx.Unlock() | ||
} | ||
|
||
func (s *streamSeriesSet) At() ([]storepb.Label, []storepb.AggrChunk) { | ||
if s.currSeries == nil { | ||
return nil, nil | ||
|
@@ -411,11 +434,14 @@ func (s *ProxyStore) LabelValues(ctx context.Context, r *storepb.LabelValuesRequ | |
g, gctx = errgroup.WithContext(ctx) | ||
) | ||
|
||
storeReadCtx, storeReadCancelFunc := s.contextWithReadTimeout(gctx) | ||
defer storeReadCancelFunc() | ||
|
||
for _, st := range s.stores() { | ||
store := st | ||
g.Go(func() error { | ||
resp, err := store.LabelValues(gctx, &storepb.LabelValuesRequest{ | ||
Label: r.Label, | ||
resp, err := store.LabelValues(storeReadCtx, &storepb.LabelValuesRequest{ | ||
Label: r.Label, | ||
PartialResponseDisabled: r.PartialResponseDisabled, | ||
}) | ||
if err != nil { | ||
|
@@ -448,3 +474,11 @@ func (s *ProxyStore) LabelValues(ctx context.Context, r *storepb.LabelValuesRequ | |
Warnings: warnings, | ||
}, nil | ||
} | ||
|
||
func (s *ProxyStore) contextWithReadTimeout(ctx context.Context) (context.Context, context.CancelFunc) { | ||
if s.readTimeout > 0 { | ||
return context.WithTimeout(ctx, s.readTimeout) | ||
} | ||
|
||
return ctx, func() {} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
missing trailing period in comment, same below