mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-27 15:30:45 +01:00
s3: keep transient object-entry errors out of ErrObjectNotFound (#12083)
* s3: keep transient object-entry errors out of ErrObjectNotFound getObjectEntry wrapped every fetch error as ErrObjectNotFound, so a transient filer failure (busy filer, ctx deadline, transport blip) was indistinguishable from a real miss. LifecycleDelete resolves ErrObjectNotFound permanently as NOOP_RESOLVED, so one flaky lookup at dispatch time silently skipped the object's expiration - errors=0, no retry, no log. This matches the flaky s3tests test_lifecyclev2_expiration signature: exactly one object of a same-rule pair surviving past its window. Only collapse genuine misses (filer not-found, version/latest-version sentinels); everything else propagates so the worker retries it as RETRY_LATER instead of resolving it. Also seed bucketConfigCache in the retention test server and teach the fake filer to inject per-path lookup errors. * s3: keep invalid version ids on the not-found path in getObjectEntry A malformed versionId used to collapse into ErrObjectNotFound like every other fetch error, so the handlers answered NoSuchKey. With transient errors now propagated, an invalid id would surface as InternalError; fold it back into the miss classification. * filer_pb: keep Unavailable out of LookupEntry's not-found normalization * s3: reject Unavailable in isObjectEntryMiss before message matching * s3: keep transient fallback errors out of latest-version resolution * filer_pb: limit LookupEntry's not-found text match to miss statuses * s3: limit isFilerNotFound text matching to miss statuses
This commit is contained in:
1 parent
8498ad716b
commit
c27d7fe5a4
5 files changed
+201
-9
No files matched your search
@@ -13,6 +13,7 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
"github.com/viant/ptrie"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/protobuf/proto"
|
||||
)
|
||||
|
||||
@@ -210,7 +211,18 @@ func UpdateEntryWithResponse(ctx context.Context, client SeaweedFilerClient, req
|
||||
func LookupEntry(ctx context.Context, client SeaweedFilerClient, request *LookupDirectoryEntryRequest) (*LookupDirectoryEntryResponse, error) {
|
||||
resp, err := client.LookupDirectoryEntry(ctx, request)
|
||||
if err != nil {
|
||||
if err == ErrNotFound || strings.Contains(err.Error(), ErrNotFound.Error()) {
|
||||
if err == ErrNotFound {
|
||||
return nil, ErrNotFound
|
||||
}
|
||||
// The filer converts its own miss to codes.NotFound; a raw sentinel
|
||||
// crossing gRPC arrives as codes.Unknown carrying only its text. The
|
||||
// same text on any other status is a failed read, not a miss.
|
||||
if st, ok := util.ServerStatus(err); ok {
|
||||
if st.Code() == codes.NotFound ||
|
||||
(st.Code() == codes.Unknown && strings.Contains(st.Message(), ErrNotFound.Error())) {
|
||||
return nil, ErrNotFound
|
||||
}
|
||||
} else if strings.Contains(err.Error(), ErrNotFound.Error()) {
|
||||
return nil, ErrNotFound
|
||||
}
|
||||
glog.V(3).InfofCtx(ctx, "read %s/%v: %v", request.Directory, request.Name, err)
|
||||
|
||||
@@ -14,8 +14,10 @@ import (
|
||||
// isFilerNotFound reports whether a filer error is a not-found.
|
||||
// Unlike lookups (normalized in filer_pb.LookupEntry), list and cache errors
|
||||
// cross gRPC as raw status errors, so the sentinel survives as codes.NotFound
|
||||
// or only as text. The text is matched last and only on what the filer itself
|
||||
// said, since callers wrap these errors with a path the client chose.
|
||||
// or as codes.Unknown carrying only its text. The text is matched only on
|
||||
// miss-shaped statuses and only on what the filer itself said: the same
|
||||
// message on another status is a failed read, and callers wrap these errors
|
||||
// with a path the client chose.
|
||||
func isFilerNotFound(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
@@ -24,7 +26,8 @@ func isFilerNotFound(err error) bool {
|
||||
return true
|
||||
}
|
||||
if st, ok := util.ServerStatus(err); ok {
|
||||
return st.Code() == codes.NotFound || strings.Contains(st.Message(), filer_pb.ErrNotFound.Error())
|
||||
return st.Code() == codes.NotFound ||
|
||||
(st.Code() == codes.Unknown && strings.Contains(st.Message(), filer_pb.ErrNotFound.Error()))
|
||||
}
|
||||
return strings.Contains(err.Error(), filer_pb.ErrNotFound.Error())
|
||||
}
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
"google.golang.org/grpc/codes"
|
||||
)
|
||||
|
||||
// ====================================================================
|
||||
@@ -208,12 +209,35 @@ func (s3a *S3ApiServer) getObjectEntry(bucket, object, versionId string) (*filer
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to retrieve object %s/%s: %w", bucket, object, ErrObjectNotFound)
|
||||
// Only a genuine miss becomes ErrObjectNotFound. Transport and
|
||||
// internal failures must stay distinguishable: callers like
|
||||
// LifecycleDelete resolve a miss permanently, so collapsing a
|
||||
// flaky read into "not found" silently drops the object.
|
||||
if isObjectEntryMiss(err) {
|
||||
return nil, fmt.Errorf("failed to retrieve object %s/%s: %w", bucket, object, ErrObjectNotFound)
|
||||
}
|
||||
return nil, fmt.Errorf("failed to retrieve object %s/%s: %w", bucket, object, err)
|
||||
}
|
||||
|
||||
return entry, nil
|
||||
}
|
||||
|
||||
// isObjectEntryMiss reports whether an object-entry fetch error is a genuine
|
||||
// miss (no object, version, or latest version) rather than a transient or
|
||||
// internal failure the caller may want to retry.
|
||||
func isObjectEntryMiss(err error) bool {
|
||||
// Unavailable is a failed read, not a miss - a wrapped not-found
|
||||
// message in its status text must not classify it as one.
|
||||
if st, ok := util.ServerStatus(err); ok && st.Code() == codes.Unavailable {
|
||||
return false
|
||||
}
|
||||
return isFilerNotFound(err) ||
|
||||
errors.Is(err, ErrObjectNotFound) ||
|
||||
errors.Is(err, ErrVersionNotFound) ||
|
||||
errors.Is(err, ErrLatestVersionNotFound) ||
|
||||
errors.Is(err, errInvalidVersionID)
|
||||
}
|
||||
|
||||
// getObjectEntryForUpdate resolves the entry a metadata update targets and the
|
||||
// filer path it is written under. It applies the same version resolution as
|
||||
// getSpecificObjectVersion - "null" names the plain pre-versioning entry, not a
|
||||
|
||||
@@ -2,6 +2,7 @@ package s3api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
@@ -11,7 +12,9 @@ import (
|
||||
"time"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
"google.golang.org/grpc/status"
|
||||
"google.golang.org/protobuf/proto"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb"
|
||||
@@ -871,14 +874,18 @@ func timePtr(t time.Time) *time.Time {
|
||||
type retentionTestFiler struct {
|
||||
filer_pb.UnimplementedSeaweedFilerServer
|
||||
|
||||
mu sync.Mutex
|
||||
entries map[string]*filer_pb.Entry
|
||||
updates []string
|
||||
mu sync.Mutex
|
||||
entries map[string]*filer_pb.Entry
|
||||
updates []string
|
||||
lookupErrs map[string]error
|
||||
}
|
||||
|
||||
func (f *retentionTestFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
if err, ok := f.lookupErrs[req.Directory+"/"+req.Name]; ok {
|
||||
return nil, err
|
||||
}
|
||||
if entry, ok := f.entries[req.Directory+"/"+req.Name]; ok {
|
||||
return &filer_pb.LookupDirectoryEntryResponse{Entry: proto.Clone(entry).(*filer_pb.Entry)}, nil
|
||||
}
|
||||
@@ -903,7 +910,8 @@ func newRetentionTestServer(t *testing.T, f *retentionTestFiler) *S3ApiServer {
|
||||
GrpcDialOption: dialOption,
|
||||
BucketsPath: "/buckets",
|
||||
},
|
||||
filerClient: wdclient.NewFilerClient([]pb.ServerAddress{addr}, dialOption, ""),
|
||||
filerClient: wdclient.NewFilerClient([]pb.ServerAddress{addr}, dialOption, ""),
|
||||
bucketConfigCache: NewBucketConfigCache(time.Minute),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -957,3 +965,140 @@ func TestNullVersionLockWritesToPlainEntry(t *testing.T) {
|
||||
t.Fatalf("plain entry legal hold = %q, want ON", got)
|
||||
}
|
||||
}
|
||||
|
||||
// getObjectEntry must only report ErrObjectNotFound for a genuine miss. A
|
||||
// transient filer error used to be collapsed into ErrObjectNotFound, which
|
||||
// LifecycleDelete resolves permanently as NOOP_RESOLVED - one flaky read at
|
||||
// dispatch time would silently skip the object's expiration.
|
||||
func TestGetObjectEntryPreservesTransientErrors(t *testing.T) {
|
||||
f := &retentionTestFiler{
|
||||
entries: map[string]*filer_pb.Entry{
|
||||
"/buckets/b": {Name: "b", IsDirectory: true},
|
||||
},
|
||||
lookupErrs: map[string]error{
|
||||
"/buckets/b/key": status.Error(codes.Unavailable, "filer overloaded"),
|
||||
},
|
||||
}
|
||||
s3a := newRetentionTestServer(t, f)
|
||||
|
||||
_, err := s3a.getObjectEntry("b", "key", "")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for transient lookup failure")
|
||||
}
|
||||
if errors.Is(err, ErrObjectNotFound) || errors.Is(err, filer_pb.ErrNotFound) {
|
||||
t.Fatalf("transient lookup error must not collapse to not-found: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// A transport failure whose status text happens to carry the filer not-found
|
||||
// sentinel (e.g. a proxy re-wrapping a miss as Unavailable) is a failed read,
|
||||
// not a miss - LookupEntry must not normalize it to ErrNotFound.
|
||||
func TestGetObjectEntryPreservesUnavailableWithNotFoundMessage(t *testing.T) {
|
||||
f := &retentionTestFiler{
|
||||
entries: map[string]*filer_pb.Entry{
|
||||
"/buckets/b": {Name: "b", IsDirectory: true},
|
||||
},
|
||||
lookupErrs: map[string]error{
|
||||
"/buckets/b/key": status.Error(codes.Unavailable, filer_pb.ErrNotFound.Error()),
|
||||
},
|
||||
}
|
||||
s3a := newRetentionTestServer(t, f)
|
||||
|
||||
_, err := s3a.getObjectEntry("b", "key", "")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for unavailable lookup")
|
||||
}
|
||||
if errors.Is(err, ErrObjectNotFound) || errors.Is(err, filer_pb.ErrNotFound) {
|
||||
t.Fatalf("unavailable lookup must not collapse to not-found: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Only miss-shaped statuses may carry the not-found text into ErrObjectNotFound:
|
||||
// a raw sentinel crossing gRPC arrives as codes.Unknown, while the same text on
|
||||
// codes.Internal is a failed read.
|
||||
func TestGetObjectEntryNotFoundTextStatusGate(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
err error
|
||||
wantMiss bool
|
||||
}{
|
||||
{"unknown keeps sentinel text", status.Error(codes.Unknown, filer_pb.ErrNotFound.Error()), true},
|
||||
{"internal with sentinel text", status.Error(codes.Internal, filer_pb.ErrNotFound.Error()), false},
|
||||
{"not-found status", status.Error(codes.NotFound, "entry gone"), true},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
f := &retentionTestFiler{
|
||||
entries: map[string]*filer_pb.Entry{
|
||||
"/buckets/b": {Name: "b", IsDirectory: true},
|
||||
},
|
||||
lookupErrs: map[string]error{
|
||||
"/buckets/b/key": tc.err,
|
||||
},
|
||||
}
|
||||
s3a := newRetentionTestServer(t, f)
|
||||
|
||||
_, err := s3a.getObjectEntry("b", "key", "")
|
||||
if err == nil {
|
||||
t.Fatal("expected error")
|
||||
}
|
||||
if got := errors.Is(err, ErrObjectNotFound); got != tc.wantMiss {
|
||||
t.Fatalf("isObjectEntryMiss = %v, want %v: %v", got, tc.wantMiss, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// A pre-versioning object under a now-versioned bucket has no .versions
|
||||
// directory; when the plain-object fallback read fails transiently, that
|
||||
// failure must surface instead of the directory's not-found error.
|
||||
func TestGetObjectEntryPreservesTransientFallbackError(t *testing.T) {
|
||||
f := &retentionTestFiler{
|
||||
entries: map[string]*filer_pb.Entry{
|
||||
"/buckets/b": {
|
||||
Name: "b",
|
||||
IsDirectory: true,
|
||||
Extended: map[string][]byte{s3_constants.ExtVersioningKey: []byte(s3_constants.VersioningEnabled)},
|
||||
},
|
||||
},
|
||||
lookupErrs: map[string]error{
|
||||
"/buckets/b/key": status.Error(codes.Unavailable, "filer overloaded"),
|
||||
},
|
||||
}
|
||||
s3a := newRetentionTestServer(t, f)
|
||||
|
||||
_, err := s3a.getObjectEntry("b", "key", "")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for transient fallback lookup failure")
|
||||
}
|
||||
if errors.Is(err, ErrObjectNotFound) || errors.Is(err, filer_pb.ErrNotFound) {
|
||||
t.Fatalf("transient fallback error must not collapse to not-found: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetObjectEntryInvalidVersionIdStillNotFound(t *testing.T) {
|
||||
f := &retentionTestFiler{
|
||||
entries: map[string]*filer_pb.Entry{
|
||||
"/buckets/b": {Name: "b", IsDirectory: true},
|
||||
},
|
||||
}
|
||||
s3a := newRetentionTestServer(t, f)
|
||||
|
||||
_, err := s3a.getObjectEntry("b", "key", "not/../a-version")
|
||||
if !errors.Is(err, ErrObjectNotFound) {
|
||||
t.Fatalf("invalid version id must report ErrObjectNotFound, got: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetObjectEntryMissStillNotFound(t *testing.T) {
|
||||
f := &retentionTestFiler{
|
||||
entries: map[string]*filer_pb.Entry{
|
||||
"/buckets/b": {Name: "b", IsDirectory: true},
|
||||
},
|
||||
}
|
||||
s3a := newRetentionTestServer(t, f)
|
||||
|
||||
_, err := s3a.getObjectEntry("b", "missing", "")
|
||||
if !errors.Is(err, ErrObjectNotFound) {
|
||||
t.Fatalf("missing object must report ErrObjectNotFound, got: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -1993,6 +1993,11 @@ func (s3a *S3ApiServer) doGetLatestObjectVersion(bucket, object string, maxRetri
|
||||
}, maxRetries)
|
||||
|
||||
if err != nil {
|
||||
// Only a genuine miss means "no .versions directory" and earns the
|
||||
// pre-versioning fallback; a failed lookup says nothing about the dir.
|
||||
if !isObjectEntryMiss(err) {
|
||||
return nil, fmt.Errorf("failed to get %s/%s .versions directory: %w", bucket, normalizedObject, err)
|
||||
}
|
||||
// .versions directory doesn't exist - this can happen for objects that existed
|
||||
// before versioning was enabled on the bucket. Fall back to checking for a
|
||||
// regular (non-versioned) object file.
|
||||
@@ -2000,6 +2005,9 @@ func (s3a *S3ApiServer) doGetLatestObjectVersion(bucket, object string, maxRetri
|
||||
|
||||
regularEntry, regularErr := s3a.getEntry(bucketDir, normalizedObject)
|
||||
if regularErr != nil {
|
||||
if !isObjectEntryMiss(regularErr) {
|
||||
return nil, fmt.Errorf("failed to get regular object %s/%s: %w", bucket, normalizedObject, regularErr)
|
||||
}
|
||||
glog.V(1).Infof("getLatestObjectVersion: no pre-versioning object found for %s/%s (error: %v)", bucket, normalizedObject, regularErr)
|
||||
return nil, fmt.Errorf("failed to get %s/%s .versions directory and no regular object found: %w", bucket, normalizedObject, err)
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user