mirrors/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2025-11-04 10:21:02 +01:00
Code Issues Projects Releases Wiki Activity

Merge pull request #11727 from prometheus/fix-error-unwrapping

Browse Source 
Correctly handle error unwrapping in rules and remote write receiver
This commit is contained in:
Julien Pivotto 2022-12-15 13:44:26 +01:00 committed by GitHub
commit 4f35683b65
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 23 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
rules/manager.go
View File

@ -673,6 +673,9 @@ func (g *Group) Eval(ctx context.Context, ts time.Time) {
rule.SetLastError(err) rule.SetLastError(err)
sp.SetStatus(codes.Error, err.Error()) sp.SetStatus(codes.Error, err.Error())
unwrappedErr := errors.Unwrap(err) unwrappedErr := errors.Unwrap(err)
if unwrappedErr == nil {
unwrappedErr = err
}
switch { switch {
case errors.Is(unwrappedErr, storage.ErrOutOfOrderSample): case errors.Is(unwrappedErr, storage.ErrOutOfOrderSample):
numOutOfOrder++ numOutOfOrder++
@ -700,6 +703,9 @@ func (g *Group) Eval(ctx context.Context, ts time.Time) {
// Series no longer exposed, mark it stale. // Series no longer exposed, mark it stale.
_, err = app.Append(0, lset, timestamp.FromTime(ts), math.Float64frombits(value.StaleNaN)) _, err = app.Append(0, lset, timestamp.FromTime(ts), math.Float64frombits(value.StaleNaN))
unwrappedErr := errors.Unwrap(err) unwrappedErr := errors.Unwrap(err)
if unwrappedErr == nil {
unwrappedErr = err
}
switch { switch {
case unwrappedErr == nil: case unwrappedErr == nil:
case errors.Is(unwrappedErr, storage.ErrOutOfOrderSample), errors.Is(unwrappedErr, storage.ErrDuplicateSampleForTimestamp): case errors.Is(unwrappedErr, storage.ErrOutOfOrderSample), errors.Is(unwrappedErr, storage.ErrDuplicateSampleForTimestamp):
@ -727,6 +733,9 @@ func (g *Group) cleanupStaleSeries(ctx context.Context, ts time.Time) {
// Rule that produced series no longer configured, mark it stale. // Rule that produced series no longer configured, mark it stale.
_, err := app.Append(0, s, timestamp.FromTime(ts), math.Float64frombits(value.StaleNaN)) _, err := app.Append(0, s, timestamp.FromTime(ts), math.Float64frombits(value.StaleNaN))
unwrappedErr := errors.Unwrap(err) unwrappedErr := errors.Unwrap(err)
if unwrappedErr == nil {
unwrappedErr = err
}
switch { switch {
case unwrappedErr == nil: case unwrappedErr == nil:
case errors.Is(unwrappedErr, storage.ErrOutOfOrderSample), errors.Is(unwrappedErr, storage.ErrDuplicateSampleForTimestamp): case errors.Is(unwrappedErr, storage.ErrOutOfOrderSample), errors.Is(unwrappedErr, storage.ErrDuplicateSampleForTimestamp):

19
storage/remote/write_handler.go
View File

@ -67,11 +67,14 @@ func (h *writeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// checkAppendExemplarError modifies the AppendExamplar's returned error based on the error cause. // checkAppendExemplarError modifies the AppendExamplar's returned error based on the error cause.
func (h *writeHandler) checkAppendExemplarError(err error, e exemplar.Exemplar, outOfOrderErrs *int) error { func (h *writeHandler) checkAppendExemplarError(err error, e exemplar.Exemplar, outOfOrderErrs *int) error {
unwrapedErr := errors.Unwrap(err) unwrappedErr := errors.Unwrap(err)
if unwrappedErr == nil {
unwrappedErr = err
}
switch { switch {
case errors.Is(unwrapedErr, storage.ErrNotFound): case errors.Is(unwrappedErr, storage.ErrNotFound):
return storage.ErrNotFound return storage.ErrNotFound
case errors.Is(unwrapedErr, storage.ErrOutOfOrderExemplar): case errors.Is(unwrappedErr, storage.ErrOutOfOrderExemplar):
*outOfOrderErrs++ *outOfOrderErrs++
level.Debug(h.logger).Log("msg", "Out of order exemplar", "exemplar", fmt.Sprintf("%+v", e)) level.Debug(h.logger).Log("msg", "Out of order exemplar", "exemplar", fmt.Sprintf("%+v", e))
return nil return nil
@ -98,8 +101,11 @@ func (h *writeHandler) write(ctx context.Context, req *prompb.WriteRequest) (err
for _, s := range ts.Samples { for _, s := range ts.Samples {
_, err = app.Append(0, labels, s.Timestamp, s.Value) _, err = app.Append(0, labels, s.Timestamp, s.Value)
if err != nil { if err != nil {
unwrapedErr := errors.Unwrap(err) unwrappedErr := errors.Unwrap(err)
if errors.Is(unwrapedErr, storage.ErrOutOfOrderSample) || errors.Is(unwrapedErr, storage.ErrOutOfBounds) || errors.Is(unwrapedErr, storage.ErrDuplicateSampleForTimestamp) { if unwrappedErr == nil {
unwrappedErr = err
}
if errors.Is(err, storage.ErrOutOfOrderSample) || errors.Is(unwrappedErr, storage.ErrOutOfBounds) || errors.Is(unwrappedErr, storage.ErrDuplicateSampleForTimestamp) {
level.Error(h.logger).Log("msg", "Out of order sample from remote write", "err", err.Error(), "series", labels.String(), "timestamp", s.Timestamp) level.Error(h.logger).Log("msg", "Out of order sample from remote write", "err", err.Error(), "series", labels.String(), "timestamp", s.Timestamp)
} }
return err return err
@ -123,6 +129,9 @@ func (h *writeHandler) write(ctx context.Context, req *prompb.WriteRequest) (err
_, err = app.AppendHistogram(0, labels, hp.Timestamp, hs) _, err = app.AppendHistogram(0, labels, hp.Timestamp, hs)
if err != nil { if err != nil {
unwrappedErr := errors.Unwrap(err) unwrappedErr := errors.Unwrap(err)
if unwrappedErr == nil {
unwrappedErr = err
}
// Althogh AppendHistogram does not currently return ErrDuplicateSampleForTimestamp there is // Althogh AppendHistogram does not currently return ErrDuplicateSampleForTimestamp there is
// a note indicating its inclusion in the future. // a note indicating its inclusion in the future.
if errors.Is(unwrappedErr, storage.ErrOutOfOrderSample) || errors.Is(unwrappedErr, storage.ErrOutOfBounds) || errors.Is(unwrappedErr, storage.ErrDuplicateSampleForTimestamp) { if errors.Is(unwrappedErr, storage.ErrOutOfOrderSample) || errors.Is(unwrappedErr, storage.ErrOutOfBounds) || errors.Is(unwrappedErr, storage.ErrDuplicateSampleForTimestamp) {