mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-23 22:40:39 +02:00
s3: surface transient store errors during multipart resume/copy instead of masking them (#10207)
* s3: surface multipart-list store errors instead of masking them
listMultipartUploads masked a filer/metadata-store list error as an empty
200 response, and listObjectParts masked it as NoSuchUpload. A client
resuming a multipart upload (the Docker Registry S3 driver resolves an
in-progress upload via ListMultipartUploads, then ListParts) reads either
as "upload gone" and fails the upload permanently with a non-retryable
error. Return ErrInternalError so a transient store error stays retryable,
keeping ErrNotFound as an empty list / NoSuchUpload respectively.
* s3: return retryable error for transient CopyObject source lookups
Resolving the copy source is reported as InvalidArgument ("Copy Source must
mention the source bucket and key") whenever the lookup returns any error,
including a transient store error. Clients don't retry a 400, so a resumable
blob commit fails permanently. Keep the client error for a missing, invalid,
directory, or delete-marker source; map a transient store error to
ErrInternalError.
* s3: mark versioned lookup miss with the not-found sentinel
recoverLatestVersionWithoutPointer's terminal miss (a .versions directory
with no pointer, no versions, and no null object) returned a plain error,
so errors.Is-based callers could not tell this designed NoSuchKey state
from a store failure and reported it as an internal error.
* s3: reject a delete-marker copy source with NoSuchKey
getLatestObjectVersion returns the delete-marker entry when the latest
version is a delete marker, and the copy handlers copied its empty stub
into the destination. Every other handler detects ExtDeleteMarkerKey and
answers NoSuchKey; do the same for CopyObject and UploadPartCopy.
* s3: align copy-source not-found detection with the grpc status idiom
The lookup path canonicalizes text-form not-found into the sentinel in
filer_pb.LookupEntry and wraps with %w after that, so matching sentinel
text here was unreachable -- and risky, since a store error whose message
merely mentions the sentinel would be downgraded to a terminal 400.
Match the raw grpc NotFound code instead, like the other version-lookup
sites, and give transient filer errors the 503 that the upload-entry
lookup in this file already returns.
* s3: keep source-bucket versioning lookup errors retryable in copy
CopyObject and UploadPartCopy mapped any source-bucket versioning-state
lookup error to a terminal InvalidCopySource, including transient store
errors; getVersioningState signals a missing bucket with the not-found
sentinel, so split on that and let real store errors surface as 500, the
same mapping the destination-bucket lookup already uses.
* s3: match grpc-transported not-found in multipart list error paths
The list client path returns raw grpc status errors without the sentinel
reconstruction that lookups get in filer_pb.LookupEntry, so errors.Is on
the not-found sentinel never matched a store-reported missing directory;
match the sentinel text as well via a shared helper. The common missing-
directory case still lists as empty with no error and is unaffected.
* s3: complete and abort multipart surface store errors
prepareMultipartCompletionState mapped any upload-directory list or
lookup error to NoSuchUpload, so a transient store error at completion
time made the client discard a fully-uploaded object as gone. Split on
not-found like the sibling listing paths. abortMultipartUpload did the
same on s3a.exists, whose errors are never not-found (filer_pb.Exists
reports that as false with no error) -- a store failure there answered
NoSuchUpload and silently leaked the uploaded parts.
* s3: reject part numbers below 1 in UploadPartCopy
Only the upper bound was checked, unlike PutObjectPart; partNumber=0
passed the route regex and validation and wrote an out-of-range
0000_copy.part into the upload directory.
This commit is contained in:
1 parent
9b1ff91949
commit
77694d3a93
5 files changed
+140
-17
No files matched your search
@@ -361,8 +361,12 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input *
|
||||
entries, _, err := s3a.list(uploadDirectory, "", "", false, s3_constants.MaxS3MultipartParts+1)
|
||||
if err != nil {
|
||||
glog.Errorf("completeMultipartUpload %s %s error: %v", *input.Bucket, *input.UploadId, err)
|
||||
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc()
|
||||
return nil, nil, s3err.ErrNoSuchUpload
|
||||
if isFilerListNotFound(err) {
|
||||
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc()
|
||||
return nil, nil, s3err.ErrNoSuchUpload
|
||||
}
|
||||
// a store error must not read as "upload gone": the client would drop a fully-uploaded object
|
||||
return nil, nil, s3err.ErrInternalError
|
||||
}
|
||||
if len(entries) == 0 {
|
||||
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc()
|
||||
@@ -372,8 +376,11 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input *
|
||||
pentry, err := s3a.getEntry(s3a.genUploadsFolder(*input.Bucket), *input.UploadId)
|
||||
if err != nil {
|
||||
glog.Errorf("completeMultipartUpload %s %s error: %v", *input.Bucket, *input.UploadId, err)
|
||||
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc()
|
||||
return nil, nil, s3err.ErrNoSuchUpload
|
||||
if errors.Is(err, filer_pb.ErrNotFound) {
|
||||
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc()
|
||||
return nil, nil, s3err.ErrNoSuchUpload
|
||||
}
|
||||
return nil, nil, s3err.ErrInternalError
|
||||
}
|
||||
|
||||
deleteEntries := make([]*filer_pb.Entry, 0)
|
||||
@@ -882,8 +889,10 @@ func (s3a *S3ApiServer) abortMultipartUpload(input *s3.AbortMultipartUploadInput
|
||||
|
||||
exists, err := s3a.exists(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true)
|
||||
if err != nil {
|
||||
glog.V(1).Infof("bucket %s abort upload %s: %v", *input.Bucket, *input.UploadId, err)
|
||||
return nil, s3err.ErrNoSuchUpload
|
||||
// filer_pb.Exists reports not-found as (false, nil), so an error here is
|
||||
// always a store failure; answering NoSuchUpload would leak the parts.
|
||||
glog.Errorf("bucket %s abort upload %s: %v", *input.Bucket, *input.UploadId, err)
|
||||
return nil, s3err.ErrInternalError
|
||||
}
|
||||
if exists {
|
||||
err = s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true, true)
|
||||
@@ -930,8 +939,14 @@ func (s3a *S3ApiServer) listMultipartUploads(input *s3.ListMultipartUploadsInput
|
||||
|
||||
entries, _, err := s3a.list(s3a.genUploadsFolder(*input.Bucket), "", *input.UploadIdMarker, false, math.MaxInt32)
|
||||
if err != nil {
|
||||
// A missing .uploads folder normally lists as empty with no error; a
|
||||
// store that reports it as not-found still means an empty list.
|
||||
if isFilerListNotFound(err) {
|
||||
return output, s3err.ErrNone
|
||||
}
|
||||
// surface a real store error; a masked empty 200 makes a resuming client treat the upload as gone
|
||||
glog.Errorf("listMultipartUploads %s error: %v", *input.Bucket, err)
|
||||
return
|
||||
return output, s3err.ErrInternalError
|
||||
}
|
||||
|
||||
uploadsCount := int64(0)
|
||||
@@ -991,8 +1006,13 @@ func (s3a *S3ApiServer) listObjectParts(input *s3.ListPartsInput) (output *ListP
|
||||
|
||||
entries, isLast, err := s3a.list(s3a.genUploadsFolder(*input.Bucket)+"/"+*input.UploadId, "", fmt.Sprintf("%04d%s", *input.PartNumberMarker, multipartExt), false, uint32(*input.MaxParts))
|
||||
if err != nil {
|
||||
// A store that reports the missing upload directory as not-found means
|
||||
// the upload is gone (completed or aborted), not a store error.
|
||||
if isFilerListNotFound(err) {
|
||||
return nil, s3err.ErrNoSuchUpload
|
||||
}
|
||||
glog.Errorf("listObjectParts %s %s error: %v", *input.Bucket, *input.UploadId, err)
|
||||
return nil, s3err.ErrNoSuchUpload
|
||||
return nil, s3err.ErrInternalError
|
||||
}
|
||||
|
||||
// Note: The upload directory is sort of a marker of the existence of an multipart upload request.
|
||||
|
||||
@@ -1,10 +1,22 @@
|
||||
package s3api
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"strings"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
|
||||
)
|
||||
|
||||
// isFilerListNotFound reports whether a filer list error is a not-found.
|
||||
// Unlike lookups (normalized in filer_pb.LookupEntry), list errors cross gRPC
|
||||
// as raw status errors, so the sentinel survives only as text. Callers must
|
||||
// pass errors from paths without user-controlled segments that could spoof it.
|
||||
func isFilerListNotFound(err error) bool {
|
||||
return errors.Is(err, filer_pb.ErrNotFound) || strings.Contains(err.Error(), filer_pb.ErrNotFound.Error())
|
||||
}
|
||||
|
||||
// ErrorHandlers provide common error handling patterns for S3 API operations
|
||||
|
||||
// handleMultipartError logs an error and returns the standard multipart error format
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
package s3api
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
func TestClassifyCopySourceError(t *testing.T) {
|
||||
deleteMarker := &filer_pb.Entry{
|
||||
Extended: map[string][]byte{s3_constants.ExtDeleteMarkerKey: []byte("true")},
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
entry *filer_pb.Entry
|
||||
err error
|
||||
want s3err.ErrorCode
|
||||
}{
|
||||
{"regular entry", &filer_pb.Entry{Name: "o"}, nil, s3err.ErrNone},
|
||||
{"nil entry", nil, nil, s3err.ErrInvalidCopySource},
|
||||
{"directory entry", &filer_pb.Entry{IsDirectory: true}, nil, s3err.ErrInvalidCopySource},
|
||||
{"delete marker entry", deleteMarker, nil, s3err.ErrNoSuchKey},
|
||||
{"not found sentinel", nil, filer_pb.ErrNotFound, s3err.ErrInvalidCopySource},
|
||||
{"wrapped not found", nil, fmt.Errorf("read %s: %w", "o", filer_pb.ErrNotFound), s3err.ErrInvalidCopySource},
|
||||
{"grpc not found", nil, status.Error(codes.NotFound, "no entry"), s3err.ErrInvalidCopySource},
|
||||
{"invalid version id", nil, errInvalidVersionID, s3err.ErrInvalidCopySource},
|
||||
{"delete marker error", nil, ErrDeleteMarker, s3err.ErrInvalidCopySource},
|
||||
{"transient unavailable", nil, status.Error(codes.Unavailable, "filer down"), s3err.ErrServiceUnavailable},
|
||||
{"store error", nil, errors.New("connection reset"), s3err.ErrInternalError},
|
||||
// A message merely mentioning the sentinel text must not downgrade a store error.
|
||||
{"sentinel text in message", nil, fmt.Errorf("peer said: %s", filer_pb.ErrNotFound.Error()), s3err.ErrInternalError},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := classifyCopySourceError(tt.entry, tt.err); got != tt.want {
|
||||
t.Errorf("classifyCopySourceError() = %v, want %v", got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -24,6 +24,8 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/security"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
"google.golang.org/protobuf/proto"
|
||||
)
|
||||
|
||||
@@ -67,6 +69,31 @@ func hasPrefixFold(s, prefix string) bool {
|
||||
return len(s) >= len(prefix) && strings.EqualFold(s[:len(prefix)], prefix)
|
||||
}
|
||||
|
||||
// classifyCopySourceError maps a copy-source lookup to an S3 error: a missing,
|
||||
// invalid, or directory source stays a client error, but a transient store
|
||||
// error becomes a retryable 5xx so a resumable copy/commit survives a blip.
|
||||
func classifyCopySourceError(entry *filer_pb.Entry, err error) s3err.ErrorCode {
|
||||
if err == nil {
|
||||
if entry == nil || entry.IsDirectory {
|
||||
return s3err.ErrInvalidCopySource
|
||||
}
|
||||
// The latest-version pointer tracks delete markers too; copying one
|
||||
// must be NoSuchKey like every other handler, not a copy of the stub.
|
||||
if deleteMarker, ok := entry.Extended[s3_constants.ExtDeleteMarkerKey]; ok && string(deleteMarker) == "true" {
|
||||
return s3err.ErrNoSuchKey
|
||||
}
|
||||
return s3err.ErrNone
|
||||
}
|
||||
if errors.Is(err, filer_pb.ErrNotFound) || status.Code(err) == codes.NotFound ||
|
||||
errors.Is(err, errInvalidVersionID) || errors.Is(err, ErrDeleteMarker) {
|
||||
return s3err.ErrInvalidCopySource
|
||||
}
|
||||
if isTransientFilerError(err) {
|
||||
return s3err.ErrServiceUnavailable
|
||||
}
|
||||
return s3err.ErrInternalError
|
||||
}
|
||||
|
||||
func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request) {
|
||||
t := time.Now().UTC().Truncate(time.Millisecond)
|
||||
|
||||
@@ -144,13 +171,19 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
|
||||
srcVersioningState, err := s3a.getVersioningState(srcBucket)
|
||||
if err != nil {
|
||||
glog.Errorf("Error checking versioning state for source bucket %s: %v", srcBucket, err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
|
||||
// Only a missing bucket is the client's fault; a store error must stay
|
||||
// retryable, matching the destination-bucket lookup.
|
||||
if errors.Is(err, filer_pb.ErrNotFound) {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
|
||||
return
|
||||
}
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
return
|
||||
}
|
||||
|
||||
entry, err := s3a.resolveCopySourceEntry(srcBucket, srcObject, srcVersionId, srcVersioningState)
|
||||
if err != nil || entry.IsDirectory {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
|
||||
if errCode := classifyCopySourceError(entry, err); errCode != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -216,8 +249,8 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
|
||||
routeInPlace := owner != "" && dstVersioningState == "" && !mimeChanged
|
||||
selfCopyBody := func() s3err.ErrorCode {
|
||||
currentEntry, currentErr := s3a.resolveCopySourceEntry(srcBucket, srcObject, srcVersionId, srcVersioningState)
|
||||
if currentErr != nil || currentEntry.IsDirectory {
|
||||
return s3err.ErrInvalidCopySource
|
||||
if errCode := classifyCopySourceError(currentEntry, currentErr); errCode != s3err.ErrNone {
|
||||
return errCode
|
||||
}
|
||||
if errCode := s3a.validateConditionalCopyHeaders(r, currentEntry); errCode != s3err.ErrNone {
|
||||
return errCode
|
||||
@@ -828,12 +861,22 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidPart)
|
||||
return
|
||||
}
|
||||
if partID < 1 {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidPart)
|
||||
return
|
||||
}
|
||||
|
||||
// Get detailed versioning state for source bucket
|
||||
srcVersioningState, err := s3a.getVersioningState(srcBucket)
|
||||
if err != nil {
|
||||
glog.Errorf("Error checking versioning state for source bucket %s: %v", srcBucket, err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
|
||||
// Only a missing bucket is the client's fault; a store error must stay
|
||||
// retryable, matching the destination-bucket lookup.
|
||||
if errors.Is(err, filer_pb.ErrNotFound) {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
|
||||
return
|
||||
}
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -856,8 +899,8 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
|
||||
entry, err = s3a.getEntry(dir, name)
|
||||
}
|
||||
|
||||
if err != nil || entry.IsDirectory {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopySource)
|
||||
if errCode := classifyCopySourceError(entry, err); errCode != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@@ -1849,7 +1849,7 @@ func (s3a *S3ApiServer) recoverLatestVersionWithoutPointer(bucket, normalizedObj
|
||||
return nil, healErr
|
||||
}
|
||||
|
||||
return nil, fmt.Errorf("no version metadata in .versions directory and no regular object found for %s/%s", bucket, normalizedObject)
|
||||
return nil, fmt.Errorf("no version metadata in .versions directory and no regular object found for %s/%s: %w", bucket, normalizedObject, filer_pb.ErrNotFound)
|
||||
}
|
||||
|
||||
// versionIdFromEntry returns a .versions child entry's version id, preferring
|
||||
|
||||
Reference in new issue
Block a user