S3: Perf related (#7463)
* reduce checks * s3 object lookup optimization * Only check versioning configuration if client requests * Consolidate SSE Entry Lookups * optimize * revert optimization for versioned objects * Removed: getObjectEntryForSSE() function * refactor * Refactoring: Added fetchObjectEntryRequired * avoid refetching * return early if not found * reuse objects from conditional check * clear cache when creating bucket
This commit is contained in:
@@ -1,11 +1,15 @@
|
|||||||
package s3api
|
package s3api
|
||||||
|
|
||||||
import "github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
|
import (
|
||||||
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||||
|
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
|
||||||
|
)
|
||||||
|
|
||||||
const s3TimeFormat = "2006-01-02T15:04:05.999Z07:00"
|
const s3TimeFormat = "2006-01-02T15:04:05.999Z07:00"
|
||||||
|
|
||||||
// ConditionalHeaderResult holds the result of conditional header checking
|
// ConditionalHeaderResult holds the result of conditional header checking
|
||||||
type ConditionalHeaderResult struct {
|
type ConditionalHeaderResult struct {
|
||||||
ErrorCode s3err.ErrorCode
|
ErrorCode s3err.ErrorCode
|
||||||
ETag string // ETag of the object (for 304 responses)
|
ETag string // ETag of the object (for 304 responses)
|
||||||
|
Entry *filer_pb.Entry // Entry fetched during conditional check (nil if not fetched or object doesn't exist)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,7 +7,6 @@ import (
|
|||||||
"encoding/xml"
|
"encoding/xml"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
|
||||||
"math"
|
"math"
|
||||||
"net/http"
|
"net/http"
|
||||||
"path"
|
"path"
|
||||||
@@ -16,6 +15,8 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||||
|
|
||||||
"github.com/aws/aws-sdk-go/private/protocol/xml/xmlutil"
|
"github.com/aws/aws-sdk-go/private/protocol/xml/xmlutil"
|
||||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3bucket"
|
"github.com/seaweedfs/seaweedfs/weed/s3api/s3bucket"
|
||||||
|
|
||||||
@@ -210,6 +211,11 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Remove bucket from negative cache after successful creation
|
||||||
|
if s3a.bucketConfigCache != nil {
|
||||||
|
s3a.bucketConfigCache.RemoveNegativeCache(bucket)
|
||||||
|
}
|
||||||
|
|
||||||
// Check for x-amz-bucket-object-lock-enabled header (S3 standard compliance)
|
// Check for x-amz-bucket-object-lock-enabled header (S3 standard compliance)
|
||||||
if objectLockHeaderValue := r.Header.Get(s3_constants.AmzBucketObjectLockEnabled); strings.EqualFold(objectLockHeaderValue, "true") {
|
if objectLockHeaderValue := r.Header.Get(s3_constants.AmzBucketObjectLockEnabled); strings.EqualFold(objectLockHeaderValue, "true") {
|
||||||
glog.V(3).Infof("PutBucketHandler: enabling Object Lock and Versioning for bucket %s due to x-amz-bucket-object-lock-enabled header", bucket)
|
glog.V(3).Infof("PutBucketHandler: enabling Object Lock and Versioning for bucket %s due to x-amz-bucket-object-lock-enabled header", bucket)
|
||||||
@@ -493,16 +499,17 @@ func (s3a *S3ApiServer) HeadBucketHandler(w http.ResponseWriter, r *http.Request
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (s3a *S3ApiServer) checkBucket(r *http.Request, bucket string) s3err.ErrorCode {
|
func (s3a *S3ApiServer) checkBucket(r *http.Request, bucket string) s3err.ErrorCode {
|
||||||
entry, err := s3a.getEntry(s3a.option.BucketsPath, bucket)
|
// Use cached bucket config instead of direct getEntry call (optimization)
|
||||||
if entry == nil || errors.Is(err, filer_pb.ErrNotFound) {
|
config, errCode := s3a.getBucketConfig(bucket)
|
||||||
return s3err.ErrNoSuchBucket
|
if errCode != s3err.ErrNone {
|
||||||
|
return errCode
|
||||||
}
|
}
|
||||||
|
|
||||||
//if iam is enabled, the access was already checked before
|
//if iam is enabled, the access was already checked before
|
||||||
if s3a.iam.isEnabled() {
|
if s3a.iam.isEnabled() {
|
||||||
return s3err.ErrNone
|
return s3err.ErrNone
|
||||||
}
|
}
|
||||||
if !s3a.hasAccess(r, entry) {
|
if !s3a.hasAccess(r, config.Entry) {
|
||||||
return s3err.ErrAccessDenied
|
return s3err.ErrAccessDenied
|
||||||
}
|
}
|
||||||
return s3err.ErrNone
|
return s3err.ErrNone
|
||||||
|
|||||||
@@ -264,6 +264,8 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request)
|
|||||||
versionId := r.URL.Query().Get("versionId")
|
versionId := r.URL.Query().Get("versionId")
|
||||||
|
|
||||||
// Check if versioning is configured for the bucket (Enabled or Suspended)
|
// Check if versioning is configured for the bucket (Enabled or Suspended)
|
||||||
|
// Note: We need to check this even if versionId is empty, because versioned buckets
|
||||||
|
// handle even "get latest version" requests differently (through .versions directory)
|
||||||
versioningConfigured, err := s3a.isVersioningConfigured(bucket)
|
versioningConfigured, err := s3a.isVersioningConfigured(bucket)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if err == filer_pb.ErrNotFound {
|
if err == filer_pb.ErrNotFound {
|
||||||
@@ -344,31 +346,47 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request)
|
|||||||
destUrl = s3a.toFilerUrl(bucket, object)
|
destUrl = s3a.toFilerUrl(bucket, object)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Check if this is a range request to an SSE object and modify the approach
|
// Fetch the correct entry for SSE processing (respects versionId)
|
||||||
|
// This consolidates entry lookups to avoid multiple filer calls
|
||||||
|
var objectEntryForSSE *filer_pb.Entry
|
||||||
originalRangeHeader := r.Header.Get("Range")
|
originalRangeHeader := r.Header.Get("Range")
|
||||||
var sseObject = false
|
var sseObject = false
|
||||||
|
|
||||||
// Pre-check if this object is SSE encrypted to avoid filer range conflicts
|
if versioningConfigured {
|
||||||
if originalRangeHeader != "" {
|
// For versioned objects, reuse the already-fetched entry
|
||||||
bucket, object := s3_constants.GetBucketAndObject(r)
|
objectEntryForSSE = entry
|
||||||
objectPath := fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object)
|
} else {
|
||||||
if objectEntry, err := s3a.getEntry("", objectPath); err == nil {
|
// For non-versioned objects, try to reuse entry from conditional header check
|
||||||
primarySSEType := s3a.detectPrimarySSEType(objectEntry)
|
if result.Entry != nil {
|
||||||
if primarySSEType == s3_constants.SSETypeC || primarySSEType == s3_constants.SSETypeKMS {
|
// Reuse entry fetched during conditional header check (optimization)
|
||||||
sseObject = true
|
objectEntryForSSE = result.Entry
|
||||||
// Temporarily remove Range header to get full encrypted data from filer
|
glog.V(3).Infof("GetObjectHandler: Reusing entry from conditional header check for %s/%s", bucket, object)
|
||||||
r.Header.Del("Range")
|
} else {
|
||||||
|
// No conditional headers were checked, fetch entry for SSE processing
|
||||||
|
var fetchErr error
|
||||||
|
objectEntryForSSE, fetchErr = s3a.fetchObjectEntry(bucket, object)
|
||||||
|
if fetchErr != nil {
|
||||||
|
glog.Errorf("GetObjectHandler: failed to get entry for SSE check: %v", fetchErr)
|
||||||
|
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if objectEntryForSSE == nil {
|
||||||
|
// Not found, return error early to avoid another lookup in proxyToFiler
|
||||||
|
s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
|
||||||
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Fetch the correct entry for SSE processing (respects versionId)
|
// Check if this is an SSE object for Range request handling
|
||||||
objectEntryForSSE, err := s3a.getObjectEntryForSSE(r, versioningConfigured, entry)
|
// This applies to both versioned and non-versioned objects
|
||||||
if err != nil {
|
if originalRangeHeader != "" && objectEntryForSSE != nil {
|
||||||
glog.Errorf("GetObjectHandler: %v", err)
|
primarySSEType := s3a.detectPrimarySSEType(objectEntryForSSE)
|
||||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
if primarySSEType == s3_constants.SSETypeC || primarySSEType == s3_constants.SSETypeKMS {
|
||||||
return
|
sseObject = true
|
||||||
|
// Temporarily remove Range header to get full encrypted data from filer
|
||||||
|
r.Header.Del("Range")
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
s3a.proxyToFiler(w, r, destUrl, false, func(proxyResponse *http.Response, w http.ResponseWriter) (statusCode int, bytesTransferred int64) {
|
s3a.proxyToFiler(w, r, destUrl, false, func(proxyResponse *http.Response, w http.ResponseWriter) (statusCode int, bytesTransferred int64) {
|
||||||
@@ -415,6 +433,8 @@ func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request
|
|||||||
versionId := r.URL.Query().Get("versionId")
|
versionId := r.URL.Query().Get("versionId")
|
||||||
|
|
||||||
// Check if versioning is configured for the bucket (Enabled or Suspended)
|
// Check if versioning is configured for the bucket (Enabled or Suspended)
|
||||||
|
// Note: We need to check this even if versionId is empty, because versioned buckets
|
||||||
|
// handle even "get latest version" requests differently (through .versions directory)
|
||||||
versioningConfigured, err := s3a.isVersioningConfigured(bucket)
|
versioningConfigured, err := s3a.isVersioningConfigured(bucket)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if err == filer_pb.ErrNotFound {
|
if err == filer_pb.ErrNotFound {
|
||||||
@@ -494,11 +514,31 @@ func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Fetch the correct entry for SSE processing (respects versionId)
|
// Fetch the correct entry for SSE processing (respects versionId)
|
||||||
objectEntryForSSE, err := s3a.getObjectEntryForSSE(r, versioningConfigured, entry)
|
// For versioned objects, reuse already-fetched entry; for non-versioned, try to reuse from conditional check
|
||||||
if err != nil {
|
var objectEntryForSSE *filer_pb.Entry
|
||||||
glog.Errorf("HeadObjectHandler: %v", err)
|
if versioningConfigured {
|
||||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
objectEntryForSSE = entry
|
||||||
return
|
} else {
|
||||||
|
// For non-versioned objects, try to reuse entry from conditional header check
|
||||||
|
if result.Entry != nil {
|
||||||
|
// Reuse entry fetched during conditional header check (optimization)
|
||||||
|
objectEntryForSSE = result.Entry
|
||||||
|
glog.V(3).Infof("HeadObjectHandler: Reusing entry from conditional header check for %s/%s", bucket, object)
|
||||||
|
} else {
|
||||||
|
// No conditional headers were checked, fetch entry for SSE processing
|
||||||
|
var fetchErr error
|
||||||
|
objectEntryForSSE, fetchErr = s3a.fetchObjectEntry(bucket, object)
|
||||||
|
if fetchErr != nil {
|
||||||
|
glog.Errorf("HeadObjectHandler: failed to get entry for SSE check: %v", fetchErr)
|
||||||
|
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if objectEntryForSSE == nil {
|
||||||
|
// Not found, return error early to avoid another lookup in proxyToFiler
|
||||||
|
s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
s3a.proxyToFiler(w, r, destUrl, false, func(proxyResponse *http.Response, w http.ResponseWriter) (statusCode int, bytesTransferred int64) {
|
s3a.proxyToFiler(w, r, destUrl, false, func(proxyResponse *http.Response, w http.ResponseWriter) (statusCode int, bytesTransferred int64) {
|
||||||
@@ -658,21 +698,27 @@ func writeFinalResponse(w http.ResponseWriter, proxyResponse *http.Response, bod
|
|||||||
return statusCode, bytesTransferred
|
return statusCode, bytesTransferred
|
||||||
}
|
}
|
||||||
|
|
||||||
// getObjectEntryForSSE fetches the correct filer entry for SSE processing
|
// fetchObjectEntry fetches the filer entry for an object
|
||||||
// For versioned objects, it reuses the already-fetched entry
|
// Returns nil if not found (not an error), or propagates other errors
|
||||||
// For non-versioned objects, it fetches the entry from the filer
|
func (s3a *S3ApiServer) fetchObjectEntry(bucket, object string) (*filer_pb.Entry, error) {
|
||||||
func (s3a *S3ApiServer) getObjectEntryForSSE(r *http.Request, versioningConfigured bool, versionedEntry *filer_pb.Entry) (*filer_pb.Entry, error) {
|
|
||||||
if versioningConfigured {
|
|
||||||
// For versioned objects, we already have the correct entry
|
|
||||||
return versionedEntry, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// For non-versioned objects, fetch the entry
|
|
||||||
bucket, object := s3_constants.GetBucketAndObject(r)
|
|
||||||
objectPath := fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object)
|
objectPath := fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object)
|
||||||
fetchedEntry, err := s3a.getEntry("", objectPath)
|
fetchedEntry, fetchErr := s3a.getEntry("", objectPath)
|
||||||
if err != nil && !errors.Is(err, filer_pb.ErrNotFound) {
|
if fetchErr != nil {
|
||||||
return nil, fmt.Errorf("failed to get entry for SSE check %s: %w", objectPath, err)
|
if errors.Is(fetchErr, filer_pb.ErrNotFound) {
|
||||||
|
return nil, nil // Not found is not an error for SSE check
|
||||||
|
}
|
||||||
|
return nil, fetchErr // Propagate other errors
|
||||||
|
}
|
||||||
|
return fetchedEntry, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// fetchObjectEntryRequired fetches the filer entry for an object
|
||||||
|
// Returns an error if the object is not found or any other error occurs
|
||||||
|
func (s3a *S3ApiServer) fetchObjectEntryRequired(bucket, object string) (*filer_pb.Entry, error) {
|
||||||
|
objectPath := fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object)
|
||||||
|
fetchedEntry, fetchErr := s3a.getEntry("", objectPath)
|
||||||
|
if fetchErr != nil {
|
||||||
|
return nil, fetchErr // Return error for both not-found and other errors
|
||||||
}
|
}
|
||||||
return fetchedEntry, nil
|
return fetchedEntry, nil
|
||||||
}
|
}
|
||||||
@@ -750,7 +796,7 @@ func (s3a *S3ApiServer) handleSSECResponse(r *http.Request, proxyResponse *http.
|
|||||||
if sseCChunks >= 1 {
|
if sseCChunks >= 1 {
|
||||||
|
|
||||||
// Handle chunked SSE-C objects - each chunk needs independent decryption
|
// Handle chunked SSE-C objects - each chunk needs independent decryption
|
||||||
multipartReader, decErr := s3a.createMultipartSSECDecryptedReader(r, proxyResponse)
|
multipartReader, decErr := s3a.createMultipartSSECDecryptedReader(r, proxyResponse, entry)
|
||||||
if decErr != nil {
|
if decErr != nil {
|
||||||
glog.Errorf("Failed to create multipart SSE-C decrypted reader: %v", decErr)
|
glog.Errorf("Failed to create multipart SSE-C decrypted reader: %v", decErr)
|
||||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||||
@@ -966,7 +1012,7 @@ func (s3a *S3ApiServer) handleSSEKMSResponse(r *http.Request, proxyResponse *htt
|
|||||||
var decryptedReader io.Reader
|
var decryptedReader io.Reader
|
||||||
if isMultipartSSEKMS {
|
if isMultipartSSEKMS {
|
||||||
// Handle multipart SSE-KMS objects - each chunk needs independent decryption
|
// Handle multipart SSE-KMS objects - each chunk needs independent decryption
|
||||||
multipartReader, decErr := s3a.createMultipartSSEKMSDecryptedReader(r, proxyResponse)
|
multipartReader, decErr := s3a.createMultipartSSEKMSDecryptedReader(r, proxyResponse, entry)
|
||||||
if decErr != nil {
|
if decErr != nil {
|
||||||
glog.Errorf("Failed to create multipart SSE-KMS decrypted reader: %v", decErr)
|
glog.Errorf("Failed to create multipart SSE-KMS decrypted reader: %v", decErr)
|
||||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||||
@@ -1271,16 +1317,8 @@ func (s3a *S3ApiServer) detectPrimarySSEType(entry *filer_pb.Entry) string {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// createMultipartSSEKMSDecryptedReader creates a reader that decrypts each chunk independently for multipart SSE-KMS objects
|
// createMultipartSSEKMSDecryptedReader creates a reader that decrypts each chunk independently for multipart SSE-KMS objects
|
||||||
func (s3a *S3ApiServer) createMultipartSSEKMSDecryptedReader(r *http.Request, proxyResponse *http.Response) (io.Reader, error) {
|
func (s3a *S3ApiServer) createMultipartSSEKMSDecryptedReader(r *http.Request, proxyResponse *http.Response, entry *filer_pb.Entry) (io.Reader, error) {
|
||||||
// Get the object path from the request
|
// Entry is passed from caller to avoid redundant filer lookup
|
||||||
bucket, object := s3_constants.GetBucketAndObject(r)
|
|
||||||
objectPath := fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object)
|
|
||||||
|
|
||||||
// Get the object entry from filer to access chunk information
|
|
||||||
entry, err := s3a.getEntry("", objectPath)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("failed to get object entry for multipart SSE-KMS decryption: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Sort chunks by offset to ensure correct order
|
// Sort chunks by offset to ensure correct order
|
||||||
chunks := entry.GetChunks()
|
chunks := entry.GetChunks()
|
||||||
@@ -1531,22 +1569,14 @@ func (r *SSERangeReader) Read(p []byte) (n int, err error) {
|
|||||||
|
|
||||||
// createMultipartSSECDecryptedReader creates a decrypted reader for multipart SSE-C objects
|
// createMultipartSSECDecryptedReader creates a decrypted reader for multipart SSE-C objects
|
||||||
// Each chunk has its own IV and encryption key from the original multipart parts
|
// Each chunk has its own IV and encryption key from the original multipart parts
|
||||||
func (s3a *S3ApiServer) createMultipartSSECDecryptedReader(r *http.Request, proxyResponse *http.Response) (io.Reader, error) {
|
func (s3a *S3ApiServer) createMultipartSSECDecryptedReader(r *http.Request, proxyResponse *http.Response, entry *filer_pb.Entry) (io.Reader, error) {
|
||||||
// Parse SSE-C headers from the request for decryption key
|
// Parse SSE-C headers from the request for decryption key
|
||||||
customerKey, err := ParseSSECHeaders(r)
|
customerKey, err := ParseSSECHeaders(r)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("invalid SSE-C headers for multipart decryption: %v", err)
|
return nil, fmt.Errorf("invalid SSE-C headers for multipart decryption: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Get the object path from the request
|
// Entry is passed from caller to avoid redundant filer lookup
|
||||||
bucket, object := s3_constants.GetBucketAndObject(r)
|
|
||||||
objectPath := fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object)
|
|
||||||
|
|
||||||
// Get the object entry from filer to access chunk information
|
|
||||||
entry, err := s3a.getEntry("", objectPath)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("failed to get object entry for multipart SSE-C decryption: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Sort chunks by offset to ensure correct order
|
// Sort chunks by offset to ensure correct order
|
||||||
chunks := entry.GetChunks()
|
chunks := entry.GetChunks()
|
||||||
|
|||||||
@@ -68,8 +68,7 @@ func (s3a *S3ApiServer) GetObjectAclHandler(w http.ResponseWriter, r *http.Reque
|
|||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
// Handle regular (non-versioned) object ACL retrieval
|
// Handle regular (non-versioned) object ACL retrieval
|
||||||
bucketDir := s3a.option.BucketsPath + "/" + bucket
|
entry, err = s3a.fetchObjectEntryRequired(bucket, object)
|
||||||
entry, err = s3a.getEntry(bucketDir, object)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if errors.Is(err, filer_pb.ErrNotFound) {
|
if errors.Is(err, filer_pb.ErrNotFound) {
|
||||||
s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
|
s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
|
||||||
@@ -212,8 +211,7 @@ func (s3a *S3ApiServer) PutObjectAclHandler(w http.ResponseWriter, r *http.Reque
|
|||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
// Handle regular (non-versioned) object ACL modification
|
// Handle regular (non-versioned) object ACL modification
|
||||||
bucketDir := s3a.option.BucketsPath + "/" + bucket
|
entry, err = s3a.fetchObjectEntryRequired(bucket, object)
|
||||||
entry, err = s3a.getEntry(bucketDir, object)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if errors.Is(err, filer_pb.ErrNotFound) {
|
if errors.Is(err, filer_pb.ErrNotFound) {
|
||||||
s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
|
s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
|
||||||
|
|||||||
@@ -1396,14 +1396,15 @@ func (s3a *S3ApiServer) checkConditionalHeadersForReadsWithGetter(getter EntryGe
|
|||||||
if !objectExists {
|
if !objectExists {
|
||||||
if headers.ifMatch != "" {
|
if headers.ifMatch != "" {
|
||||||
glog.V(3).Infof("checkConditionalHeadersForReads: If-Match failed - object %s/%s does not exist", bucket, object)
|
glog.V(3).Infof("checkConditionalHeadersForReads: If-Match failed - object %s/%s does not exist", bucket, object)
|
||||||
return ConditionalHeaderResult{ErrorCode: s3err.ErrPreconditionFailed}
|
return ConditionalHeaderResult{ErrorCode: s3err.ErrPreconditionFailed, Entry: nil}
|
||||||
}
|
}
|
||||||
if !headers.ifUnmodifiedSince.IsZero() {
|
if !headers.ifUnmodifiedSince.IsZero() {
|
||||||
glog.V(3).Infof("checkConditionalHeadersForReads: If-Unmodified-Since failed - object %s/%s does not exist", bucket, object)
|
glog.V(3).Infof("checkConditionalHeadersForReads: If-Unmodified-Since failed - object %s/%s does not exist", bucket, object)
|
||||||
return ConditionalHeaderResult{ErrorCode: s3err.ErrPreconditionFailed}
|
return ConditionalHeaderResult{ErrorCode: s3err.ErrPreconditionFailed, Entry: nil}
|
||||||
}
|
}
|
||||||
// If-None-Match and If-Modified-Since succeed when object doesn't exist
|
// If-None-Match and If-Modified-Since succeed when object doesn't exist
|
||||||
return ConditionalHeaderResult{ErrorCode: s3err.ErrNone}
|
// No entry to return since object doesn't exist
|
||||||
|
return ConditionalHeaderResult{ErrorCode: s3err.ErrNone, Entry: nil}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Object exists - check all conditions
|
// Object exists - check all conditions
|
||||||
@@ -1419,7 +1420,7 @@ func (s3a *S3ApiServer) checkConditionalHeadersForReadsWithGetter(getter EntryGe
|
|||||||
// Use production etagMatches method
|
// Use production etagMatches method
|
||||||
if !s3a.etagMatches(headers.ifMatch, objectETag) {
|
if !s3a.etagMatches(headers.ifMatch, objectETag) {
|
||||||
glog.V(3).Infof("checkConditionalHeadersForReads: If-Match failed for object %s/%s - expected ETag %s, got %s", bucket, object, headers.ifMatch, objectETag)
|
glog.V(3).Infof("checkConditionalHeadersForReads: If-Match failed for object %s/%s - expected ETag %s, got %s", bucket, object, headers.ifMatch, objectETag)
|
||||||
return ConditionalHeaderResult{ErrorCode: s3err.ErrPreconditionFailed}
|
return ConditionalHeaderResult{ErrorCode: s3err.ErrPreconditionFailed, Entry: entry}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
glog.V(3).Infof("checkConditionalHeadersForReads: If-Match passed for object %s/%s", bucket, object)
|
glog.V(3).Infof("checkConditionalHeadersForReads: If-Match passed for object %s/%s", bucket, object)
|
||||||
@@ -1430,7 +1431,7 @@ func (s3a *S3ApiServer) checkConditionalHeadersForReadsWithGetter(getter EntryGe
|
|||||||
objectModTime := time.Unix(entry.Attributes.Mtime, 0)
|
objectModTime := time.Unix(entry.Attributes.Mtime, 0)
|
||||||
if objectModTime.After(headers.ifUnmodifiedSince) {
|
if objectModTime.After(headers.ifUnmodifiedSince) {
|
||||||
glog.V(3).Infof("checkConditionalHeadersForReads: If-Unmodified-Since failed - object modified after %s", r.Header.Get(s3_constants.IfUnmodifiedSince))
|
glog.V(3).Infof("checkConditionalHeadersForReads: If-Unmodified-Since failed - object modified after %s", r.Header.Get(s3_constants.IfUnmodifiedSince))
|
||||||
return ConditionalHeaderResult{ErrorCode: s3err.ErrPreconditionFailed}
|
return ConditionalHeaderResult{ErrorCode: s3err.ErrPreconditionFailed, Entry: entry}
|
||||||
}
|
}
|
||||||
glog.V(3).Infof("checkConditionalHeadersForReads: If-Unmodified-Since passed - object not modified since %s", r.Header.Get(s3_constants.IfUnmodifiedSince))
|
glog.V(3).Infof("checkConditionalHeadersForReads: If-Unmodified-Since passed - object not modified since %s", r.Header.Get(s3_constants.IfUnmodifiedSince))
|
||||||
}
|
}
|
||||||
@@ -1442,12 +1443,12 @@ func (s3a *S3ApiServer) checkConditionalHeadersForReadsWithGetter(getter EntryGe
|
|||||||
|
|
||||||
if headers.ifNoneMatch == "*" {
|
if headers.ifNoneMatch == "*" {
|
||||||
glog.V(3).Infof("checkConditionalHeadersForReads: If-None-Match=* failed - object %s/%s exists", bucket, object)
|
glog.V(3).Infof("checkConditionalHeadersForReads: If-None-Match=* failed - object %s/%s exists", bucket, object)
|
||||||
return ConditionalHeaderResult{ErrorCode: s3err.ErrNotModified, ETag: objectETag}
|
return ConditionalHeaderResult{ErrorCode: s3err.ErrNotModified, ETag: objectETag, Entry: entry}
|
||||||
}
|
}
|
||||||
// Use production etagMatches method
|
// Use production etagMatches method
|
||||||
if s3a.etagMatches(headers.ifNoneMatch, objectETag) {
|
if s3a.etagMatches(headers.ifNoneMatch, objectETag) {
|
||||||
glog.V(3).Infof("checkConditionalHeadersForReads: If-None-Match failed - ETag matches %s", objectETag)
|
glog.V(3).Infof("checkConditionalHeadersForReads: If-None-Match failed - ETag matches %s", objectETag)
|
||||||
return ConditionalHeaderResult{ErrorCode: s3err.ErrNotModified, ETag: objectETag}
|
return ConditionalHeaderResult{ErrorCode: s3err.ErrNotModified, ETag: objectETag, Entry: entry}
|
||||||
}
|
}
|
||||||
glog.V(3).Infof("checkConditionalHeadersForReads: If-None-Match passed - ETag %s doesn't match %s", objectETag, headers.ifNoneMatch)
|
glog.V(3).Infof("checkConditionalHeadersForReads: If-None-Match passed - ETag %s doesn't match %s", objectETag, headers.ifNoneMatch)
|
||||||
}
|
}
|
||||||
@@ -1459,12 +1460,13 @@ func (s3a *S3ApiServer) checkConditionalHeadersForReadsWithGetter(getter EntryGe
|
|||||||
// Use production getObjectETag method
|
// Use production getObjectETag method
|
||||||
objectETag := s3a.getObjectETag(entry)
|
objectETag := s3a.getObjectETag(entry)
|
||||||
glog.V(3).Infof("checkConditionalHeadersForReads: If-Modified-Since failed - object not modified since %s", r.Header.Get(s3_constants.IfModifiedSince))
|
glog.V(3).Infof("checkConditionalHeadersForReads: If-Modified-Since failed - object not modified since %s", r.Header.Get(s3_constants.IfModifiedSince))
|
||||||
return ConditionalHeaderResult{ErrorCode: s3err.ErrNotModified, ETag: objectETag}
|
return ConditionalHeaderResult{ErrorCode: s3err.ErrNotModified, ETag: objectETag, Entry: entry}
|
||||||
}
|
}
|
||||||
glog.V(3).Infof("checkConditionalHeadersForReads: If-Modified-Since passed - object modified after %s", r.Header.Get(s3_constants.IfModifiedSince))
|
glog.V(3).Infof("checkConditionalHeadersForReads: If-Modified-Since passed - object modified after %s", r.Header.Get(s3_constants.IfModifiedSince))
|
||||||
}
|
}
|
||||||
|
|
||||||
return ConditionalHeaderResult{ErrorCode: s3err.ErrNone}
|
// Return success with the fetched entry for reuse
|
||||||
|
return ConditionalHeaderResult{ErrorCode: s3err.ErrNone, Entry: entry}
|
||||||
}
|
}
|
||||||
|
|
||||||
// checkConditionalHeadersForReads is the production method that uses the S3ApiServer as EntryGetter
|
// checkConditionalHeadersForReads is the production method that uses the S3ApiServer as EntryGetter
|
||||||
|
|||||||
@@ -200,8 +200,7 @@ func (s3a *S3ApiServer) getObjectEntry(bucket, object, versionId string) (*filer
|
|||||||
if versioningEnabled {
|
if versioningEnabled {
|
||||||
entry, err = s3a.getLatestObjectVersion(bucket, object)
|
entry, err = s3a.getLatestObjectVersion(bucket, object)
|
||||||
} else {
|
} else {
|
||||||
bucketDir := s3a.option.BucketsPath + "/" + bucket
|
entry, err = s3a.fetchObjectEntryRequired(bucket, object)
|
||||||
entry, err = s3a.getEntry(bucketDir, object)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -284,8 +283,7 @@ func (s3a *S3ApiServer) setObjectRetention(bucket, object, versionId string, ret
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
bucketDir := s3a.option.BucketsPath + "/" + bucket
|
entry, err = s3a.fetchObjectEntryRequired(bucket, object)
|
||||||
entry, err = s3a.getEntry(bucketDir, object)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to get object %s/%s: %w", bucket, object, ErrObjectNotFound)
|
return fmt.Errorf("failed to get object %s/%s: %w", bucket, object, ErrObjectNotFound)
|
||||||
}
|
}
|
||||||
@@ -426,8 +424,7 @@ func (s3a *S3ApiServer) setObjectLegalHold(bucket, object, versionId string, leg
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
bucketDir := s3a.option.BucketsPath + "/" + bucket
|
entry, err = s3a.fetchObjectEntryRequired(bucket, object)
|
||||||
entry, err = s3a.getEntry(bucketDir, object)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to get object %s/%s: %w", bucket, object, ErrObjectNotFound)
|
return fmt.Errorf("failed to get object %s/%s: %w", bucket, object, ErrObjectNotFound)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -165,16 +165,8 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr
|
|||||||
toWrite := min(count, int(offset+size-x))
|
toWrite := min(count, int(offset+size-x))
|
||||||
if toWrite > 0 {
|
if toWrite > 0 {
|
||||||
crc = crc.Update(buf[0:toWrite])
|
crc = crc.Update(buf[0:toWrite])
|
||||||
// the crc.Value() function is to be deprecated. this double checking is for backward compatibility
|
// Note: CRC validation happens after the loop completes (see below)
|
||||||
// with seaweed version using crc.Value() instead of uint32(crc), which appears in commit 056c480eb
|
// to avoid performance overhead in the hot read path
|
||||||
// and switch appeared in version 3.09.
|
|
||||||
if offset == 0 && size == int64(n.DataSize) && int64(count) == size && (n.Checksum != crc && uint32(n.Checksum) != crc.Value()) {
|
|
||||||
// This check works only if the buffer is big enough to hold the whole needle data
|
|
||||||
// and we ask for all needle data.
|
|
||||||
// Otherwise we cannot check the validity of partially aquired data.
|
|
||||||
stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorCRC).Inc()
|
|
||||||
return fmt.Errorf("ReadNeedleData checksum %v expected %v for Needle: %v,%v", crc, n.Checksum, v.Id, n)
|
|
||||||
}
|
|
||||||
if _, err = writer.Write(buf[0:toWrite]); err != nil {
|
if _, err = writer.Write(buf[0:toWrite]); err != nil {
|
||||||
return fmt.Errorf("ReadNeedleData write: %w", err)
|
return fmt.Errorf("ReadNeedleData write: %w", err)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user