From 4e661e2d0ce4edd68ca9a466fc4328e945743a28 Mon Sep 17 00:00:00 2001 From: Joel Verezhak Date: Fri, 27 Jun 2025 17:37:11 +0200 Subject: [PATCH 1/2] [thanos] 5407: fix nested errors --- pkg/receive/handler.go | 82 +++++++++++++++++++++++++++++++---- pkg/receive/handler_test.go | 86 ++++++++++++++++++++++++++++++++++++- 2 files changed, 159 insertions(+), 9 deletions(-) diff --git a/pkg/receive/handler.go b/pkg/receive/handler.go index 44b5fba3670..62b35687ab5 100644 --- a/pkg/receive/handler.go +++ b/pkg/receive/handler.go @@ -1096,11 +1096,40 @@ func isConflict(err error) bool { if err == nil { return false } - return err == errConflict || + + // Check direct error matches + if err == errConflict || isSampleConflictErr(err) || isExemplarConflictErr(err) || - isLabelsConflictErr(err) || - status.Code(err) == codes.AlreadyExists + isLabelsConflictErr(err) { + return true + } + + // Check gRPC status code - this handles the primary case from the issue + if status.Code(err) == codes.AlreadyExists { + return true + } + + // For wrapped errors, we need to check if the underlying cause has AlreadyExists status + // This handles the case where AlreadyExists errors are wrapped multiple times + cause := errors.Cause(err) + if status.Code(cause) == codes.AlreadyExists { + return true + } + + // Additional check for MultiError - if any error in the chain is a conflict, consider it a conflict + // This handles the nested MultiError case mentioned in the issue + if merr, ok := err.(interface{ Unwrap() []error }); ok { + if errs := merr.Unwrap(); errs != nil { + for _, e := range errs { + if isConflict(e) { + return true + } + } + } + } + + return false } // isSampleConflictErr returns whether or not the given error represents @@ -1130,15 +1159,52 @@ func isLabelsConflictErr(err error) bool { // isNotReady returns whether or not the given error represents a not ready error. func isNotReady(err error) bool { - return err == errNotReady || - err == tsdb.ErrNotReady || - status.Code(err) == codes.Unavailable + if err == nil { + return false + } + + // Check direct error matches and use errors.Is for wrapped errors + if errors.Is(err, errNotReady) || errors.Is(err, tsdb.ErrNotReady) { + return true + } + + // Check gRPC status code + if status.Code(err) == codes.Unavailable { + return true + } + + // For wrapped errors, check if the underlying cause has the correct status + cause := errors.Cause(err) + if status.Code(cause) == codes.Unavailable { + return true + } + + return false } // isUnavailable returns whether or not the given error represents an unavailable error. func isUnavailable(err error) bool { - return err == errUnavailable || - status.Code(err) == codes.Unavailable + if err == nil { + return false + } + + // Check direct error matches and use errors.Is for wrapped errors + if errors.Is(err, errUnavailable) { + return true + } + + // Check gRPC status code + if status.Code(err) == codes.Unavailable { + return true + } + + // For wrapped errors, check if the underlying cause has the correct status + cause := errors.Cause(err) + if status.Code(cause) == codes.Unavailable { + return true + } + + return false } // retryState encapsulates the number of request attempt made against a peer and, diff --git a/pkg/receive/handler_test.go b/pkg/receive/handler_test.go index e49a3f28c2a..c14c8431ad2 100644 --- a/pkg/receive/handler_test.go +++ b/pkg/receive/handler_test.go @@ -42,6 +42,8 @@ import ( "github.com/prometheus/prometheus/tsdb" "github.com/stretchr/testify/require" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "google.golang.org/grpc/test/bufconn" "github.com/thanos-io/thanos/pkg/block/metadata" @@ -660,7 +662,7 @@ func testReceiveQuorum(t *testing.T, hashringAlgo HashringAlgorithm, withConsist appender: newFakeAppender(nil, nil, nil), }, { - appender: newFakeAppender(nil, nil, nil), + appender: newFakeAppender(nil, nil, nil), }, }, }, @@ -1955,3 +1957,85 @@ func TestHandlerFlippingHashrings(t *testing.T) { cancel() wg.Wait() } + +func TestIsConflictWrappedErrors(t *testing.T) { + tests := []struct { + name string + err error + expectMatch bool + }{ + { + name: "direct AlreadyExists gRPC error", + err: status.Error(codes.AlreadyExists, "already exists"), + expectMatch: true, + }, + { + name: "wrapped AlreadyExists gRPC error", + err: errors.Wrap(status.Error(codes.AlreadyExists, "already exists"), "wrapped error"), + expectMatch: true, + }, + { + name: "double wrapped AlreadyExists gRPC error", + err: errors.Wrap(errors.Wrap(status.Error(codes.AlreadyExists, "already exists"), "first wrap"), "second wrap"), + expectMatch: true, + }, + { + name: "triple wrapped AlreadyExists gRPC error (as seen in issue #5407)", + err: errors.Wrapf( + errors.Wrapf( + status.Error(codes.AlreadyExists, "store locally for endpoint conflict"), + "forwarding request to endpoint %v", "test-endpoint"), + "replicate write request for endpoint %v", "test-endpoint"), + expectMatch: true, + }, + { + name: "non-conflict error", + err: status.Error(codes.Internal, "internal error"), + expectMatch: false, + }, + { + name: "wrapped non-conflict error", + err: errors.Wrap(status.Error(codes.Internal, "internal error"), "wrapped"), + expectMatch: false, + }, + { + name: "nil error", + err: nil, + expectMatch: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := isConflict(tt.err) + if result != tt.expectMatch { + t.Errorf("isConflict(%v) = %v, want %v", tt.err, result, tt.expectMatch) + } + }) + } +} + +// TestReplicationErrorsWithWrappedConflicts tests that replicationErrors +// properly identifies wrapped conflict errors and returns the correct cause +func TestReplicationErrorsWithWrappedConflicts(t *testing.T) { + // Create a scenario similar to the one described in issue #5407 + // where we have replication factor 2, and one request fails with wrapped AlreadyExists + re := &replicationErrors{ + threshold: 1, // With RF=2, quorum is 1, so threshold should be 1 + } + + // Add a wrapped AlreadyExists error (as would happen in the fanout scenario) + wrappedConflictErr := errors.Wrapf( + errors.Wrapf( + status.Error(codes.AlreadyExists, "store locally for endpoint conflict"), + "forwarding request to endpoint %v", "test-endpoint"), + "replicate write request for endpoint %v", "test-endpoint") + + re.Add(wrappedConflictErr) + + // The Cause() method should return errConflict, not errInternal + cause := re.Cause() + if cause != errConflict { + t.Errorf("replicationErrors.Cause() = %v, want %v", cause, errConflict) + } +} From f362d4bf1bc7206b1dc5cf4d7790175a9a94b056 Mon Sep 17 00:00:00 2001 From: Joel Verezhak Date: Fri, 27 Jun 2025 17:40:29 +0200 Subject: [PATCH 2/2] fmt --- pkg/receive/handler_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/receive/handler_test.go b/pkg/receive/handler_test.go index c14c8431ad2..3c63fd5a9a9 100644 --- a/pkg/receive/handler_test.go +++ b/pkg/receive/handler_test.go @@ -662,7 +662,7 @@ func testReceiveQuorum(t *testing.T, hashringAlgo HashringAlgorithm, withConsist appender: newFakeAppender(nil, nil, nil), }, { - appender: newFakeAppender(nil, nil, nil), + appender: newFakeAppender(nil, nil, nil), }, }, },