mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-06 06:22:05 +02:00
chrislu
4e5e8db431
Optimize DELETE and PUT operations by skipping the object lock protection check for buckets that don't have object lock enabled. Previously, enforceObjectLockProtections was always called, which performed an expensive entry lookup even for buckets without object lock. This optimization: - Adds isObjectLockEnabled() helper that uses cached bucket config - Skips enforceObjectLockProtections for buckets without object lock - Applies to both versioned and non-versioned DELETE paths - Applies to PUT operations on non-versioned buckets Performance impact: For buckets without object lock (the common case), this eliminates one filer lookup per DELETE/PUT operation.
393 lines
14 KiB
Go
393 lines
14 KiB
Go
package s3api
|
|
|
|
import (
|
|
"encoding/xml"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"strings"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/filer"
|
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
|
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
|
|
stats_collect "github.com/seaweedfs/seaweedfs/weed/stats"
|
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
|
)
|
|
|
|
const (
|
|
deleteMultipleObjectsLimit = 1000
|
|
)
|
|
|
|
func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Request) {
|
|
|
|
bucket, object := s3_constants.GetBucketAndObject(r)
|
|
glog.V(3).Infof("DeleteObjectHandler %s %s", bucket, object)
|
|
|
|
// Check for specific version ID in query parameters
|
|
versionId := r.URL.Query().Get("versionId")
|
|
|
|
// Get detailed versioning state for proper handling of suspended vs enabled versioning
|
|
versioningState, err := s3a.getVersioningState(bucket)
|
|
if err != nil {
|
|
if err == filer_pb.ErrNotFound {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket)
|
|
return
|
|
}
|
|
glog.Errorf("Error checking versioning status for bucket %s: %v", bucket, err)
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
|
return
|
|
}
|
|
|
|
versioningEnabled := (versioningState == s3_constants.VersioningEnabled)
|
|
versioningSuspended := (versioningState == s3_constants.VersioningSuspended)
|
|
versioningConfigured := (versioningState != "")
|
|
|
|
var auditLog *s3err.AccessLog
|
|
if s3err.Logger != nil {
|
|
auditLog = s3err.GetAccessLog(r, http.StatusNoContent, s3err.ErrNone)
|
|
}
|
|
|
|
if versioningConfigured {
|
|
// Handle versioned delete based on specific versioning state
|
|
if versionId != "" {
|
|
// Delete specific version (same for both enabled and suspended)
|
|
// Only check object lock if enabled for this bucket (avoids expensive entry lookup)
|
|
var prefetchedEntry *filer_pb.Entry
|
|
if s3a.isObjectLockEnabled(bucket) {
|
|
governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
|
|
var lockErr error
|
|
prefetchedEntry, lockErr = s3a.enforceObjectLockProtections(r, bucket, object, versionId, governanceBypassAllowed)
|
|
if lockErr != nil {
|
|
glog.V(2).Infof("DeleteObjectHandler: object lock check failed for %s/%s: %v", bucket, object, lockErr)
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
|
|
return
|
|
}
|
|
}
|
|
|
|
// Delete specific version, passing prefetched entry if available
|
|
err := s3a.deleteSpecificObjectVersion(bucket, object, versionId, prefetchedEntry)
|
|
if err != nil {
|
|
glog.Errorf("Failed to delete specific version %s: %v", versionId, err)
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
|
return
|
|
}
|
|
|
|
// Set version ID in response header
|
|
w.Header().Set("x-amz-version-id", versionId)
|
|
} else {
|
|
// Delete without version ID - behavior depends on versioning state
|
|
if versioningEnabled {
|
|
// Enabled versioning: Create delete marker (logical delete)
|
|
// AWS S3 behavior: Delete marker creation is NOT blocked by object retention
|
|
// because it's a logical delete that doesn't actually remove the retained version
|
|
deleteMarkerVersionId, err := s3a.createDeleteMarker(bucket, object)
|
|
if err != nil {
|
|
glog.Errorf("Failed to create delete marker: %v", err)
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
|
return
|
|
}
|
|
|
|
// Set delete marker version ID in response header
|
|
w.Header().Set("x-amz-version-id", deleteMarkerVersionId)
|
|
w.Header().Set("x-amz-delete-marker", "true")
|
|
} else if versioningSuspended {
|
|
// Suspended versioning: Actually delete the "null" version object
|
|
glog.V(2).Infof("DeleteObjectHandler: deleting null version for suspended versioning %s/%s", bucket, object)
|
|
|
|
// Only check object lock if enabled for this bucket (avoids expensive entry lookup)
|
|
var prefetchedEntry *filer_pb.Entry
|
|
if s3a.isObjectLockEnabled(bucket) {
|
|
governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
|
|
var lockErr error
|
|
prefetchedEntry, lockErr = s3a.enforceObjectLockProtections(r, bucket, object, "null", governanceBypassAllowed)
|
|
if lockErr != nil {
|
|
glog.V(2).Infof("DeleteObjectHandler: object lock check failed for %s/%s: %v", bucket, object, lockErr)
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
|
|
return
|
|
}
|
|
}
|
|
|
|
// Delete the "null" version (the regular file)
|
|
err := s3a.deleteSpecificObjectVersion(bucket, object, "null", prefetchedEntry)
|
|
if err != nil {
|
|
glog.Errorf("Failed to delete null version: %v", err)
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
|
return
|
|
}
|
|
|
|
// Note: According to AWS S3 spec, suspended versioning should NOT return version ID headers
|
|
// The object is deleted but no version information is returned
|
|
}
|
|
}
|
|
} else {
|
|
// Handle regular delete (non-versioned)
|
|
// Only check object lock permissions if object lock is enabled for the bucket
|
|
// This avoids an expensive entry lookup for buckets without object lock
|
|
if s3a.isObjectLockEnabled(bucket) {
|
|
governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
|
|
if _, lockErr := s3a.enforceObjectLockProtections(r, bucket, object, "", governanceBypassAllowed); lockErr != nil {
|
|
glog.V(2).Infof("DeleteObjectHandler: object lock check failed for %s/%s: %v", bucket, object, lockErr)
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
|
|
return
|
|
}
|
|
}
|
|
|
|
target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
|
|
dir, name := target.DirAndName()
|
|
|
|
err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
|
|
return doDeleteEntry(client, dir, name, true, false)
|
|
// Note: Empty folder cleanup is now handled asynchronously by EmptyFolderCleaner
|
|
// which listens to metadata events and uses consistent hashing for coordination
|
|
})
|
|
if err != nil {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
|
return
|
|
}
|
|
}
|
|
|
|
if auditLog != nil {
|
|
auditLog.Key = strings.TrimPrefix(object, "/")
|
|
s3err.PostAccessLog(*auditLog)
|
|
}
|
|
|
|
stats_collect.RecordBucketActiveTime(bucket)
|
|
stats_collect.S3DeletedObjectsCounter.WithLabelValues(bucket).Inc()
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
// ObjectIdentifier represents an object to be deleted with its key name and optional version ID.
|
|
type ObjectIdentifier struct {
|
|
Key string `xml:"Key"`
|
|
VersionId string `xml:"VersionId,omitempty"`
|
|
DeleteMarker bool `xml:"DeleteMarker,omitempty"`
|
|
DeleteMarkerVersionId string `xml:"DeleteMarkerVersionId,omitempty"`
|
|
}
|
|
|
|
// DeleteObjectsRequest - xml carrying the object key names which needs to be deleted.
|
|
type DeleteObjectsRequest struct {
|
|
// Element to enable quiet mode for the request
|
|
Quiet bool
|
|
// List of objects to be deleted
|
|
Objects []ObjectIdentifier `xml:"Object"`
|
|
}
|
|
|
|
// DeleteError structure.
|
|
type DeleteError struct {
|
|
Code string `xml:"Code"`
|
|
Message string `xml:"Message"`
|
|
Key string `xml:"Key"`
|
|
VersionId string `xml:"VersionId,omitempty"`
|
|
}
|
|
|
|
// DeleteObjectsResponse container for multiple object deletes.
|
|
type DeleteObjectsResponse struct {
|
|
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ DeleteResult" json:"-"`
|
|
|
|
// Collection of all deleted objects
|
|
DeletedObjects []ObjectIdentifier `xml:"Deleted,omitempty"`
|
|
|
|
// Collection of errors deleting certain objects.
|
|
Errors []DeleteError `xml:"Error,omitempty"`
|
|
}
|
|
|
|
// DeleteMultipleObjectsHandler - Delete multiple objects
|
|
func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *http.Request) {
|
|
|
|
bucket, _ := s3_constants.GetBucketAndObject(r)
|
|
glog.V(3).Infof("DeleteMultipleObjectsHandler %s", bucket)
|
|
|
|
deleteXMLBytes, err := io.ReadAll(r.Body)
|
|
if err != nil {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
|
return
|
|
}
|
|
|
|
deleteObjects := &DeleteObjectsRequest{}
|
|
if err := xml.Unmarshal(deleteXMLBytes, deleteObjects); err != nil {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrMalformedXML)
|
|
return
|
|
}
|
|
|
|
if len(deleteObjects.Objects) > deleteMultipleObjectsLimit {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidMaxDeleteObjects)
|
|
return
|
|
}
|
|
|
|
var deletedObjects []ObjectIdentifier
|
|
var deleteErrors []DeleteError
|
|
var auditLog *s3err.AccessLog
|
|
|
|
if s3err.Logger != nil {
|
|
auditLog = s3err.GetAccessLog(r, http.StatusNoContent, s3err.ErrNone)
|
|
}
|
|
|
|
// Get detailed versioning state for proper handling of suspended vs enabled versioning
|
|
versioningState, err := s3a.getVersioningState(bucket)
|
|
if err != nil {
|
|
if err == filer_pb.ErrNotFound {
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket)
|
|
return
|
|
}
|
|
glog.Errorf("Error checking versioning status for bucket %s: %v", bucket, err)
|
|
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
|
return
|
|
}
|
|
|
|
versioningEnabled := (versioningState == s3_constants.VersioningEnabled)
|
|
versioningSuspended := (versioningState == s3_constants.VersioningSuspended)
|
|
versioningConfigured := (versioningState != "")
|
|
|
|
s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
|
|
// delete file entries
|
|
for _, object := range deleteObjects.Objects {
|
|
if object.Key == "" {
|
|
continue
|
|
}
|
|
|
|
// Only check object lock if enabled for this bucket (avoids expensive entry lookup)
|
|
var prefetchedEntry *filer_pb.Entry
|
|
if versioningConfigured && s3a.isObjectLockEnabled(bucket) {
|
|
// Validate governance bypass for this specific object
|
|
governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object.Key)
|
|
var lockErr error
|
|
prefetchedEntry, lockErr = s3a.enforceObjectLockProtections(r, bucket, object.Key, object.VersionId, governanceBypassAllowed)
|
|
if lockErr != nil {
|
|
glog.V(2).Infof("DeleteMultipleObjectsHandler: object lock check failed for %s/%s (version: %s): %v", bucket, object.Key, object.VersionId, lockErr)
|
|
deleteErrors = append(deleteErrors, DeleteError{
|
|
Code: s3err.GetAPIError(s3err.ErrAccessDenied).Code,
|
|
Message: s3err.GetAPIError(s3err.ErrAccessDenied).Description,
|
|
Key: object.Key,
|
|
VersionId: object.VersionId,
|
|
})
|
|
continue
|
|
}
|
|
}
|
|
|
|
var deleteVersionId string
|
|
var isDeleteMarker bool
|
|
|
|
if versioningConfigured {
|
|
// Handle versioned delete based on specific versioning state
|
|
if object.VersionId != "" {
|
|
// Delete specific version (same for both enabled and suspended)
|
|
// Pass prefetched entry to avoid duplicate lookup
|
|
err := s3a.deleteSpecificObjectVersion(bucket, object.Key, object.VersionId, prefetchedEntry)
|
|
if err != nil {
|
|
deleteErrors = append(deleteErrors, DeleteError{
|
|
Code: "",
|
|
Message: err.Error(),
|
|
Key: object.Key,
|
|
VersionId: object.VersionId,
|
|
})
|
|
continue
|
|
}
|
|
deleteVersionId = object.VersionId
|
|
} else {
|
|
// Delete without version ID - behavior depends on versioning state
|
|
if versioningEnabled {
|
|
// Enabled versioning: Create delete marker (logical delete)
|
|
deleteMarkerVersionId, err := s3a.createDeleteMarker(bucket, object.Key)
|
|
if err != nil {
|
|
deleteErrors = append(deleteErrors, DeleteError{
|
|
Code: "",
|
|
Message: err.Error(),
|
|
Key: object.Key,
|
|
VersionId: object.VersionId,
|
|
})
|
|
continue
|
|
}
|
|
deleteVersionId = deleteMarkerVersionId
|
|
isDeleteMarker = true
|
|
} else if versioningSuspended {
|
|
// Suspended versioning: Actually delete the "null" version object
|
|
glog.V(2).Infof("DeleteMultipleObjectsHandler: deleting null version for suspended versioning %s/%s", bucket, object.Key)
|
|
|
|
// Pass prefetched entry to avoid duplicate lookup
|
|
err := s3a.deleteSpecificObjectVersion(bucket, object.Key, "null", prefetchedEntry)
|
|
if err != nil {
|
|
deleteErrors = append(deleteErrors, DeleteError{
|
|
Code: "",
|
|
Message: err.Error(),
|
|
Key: object.Key,
|
|
VersionId: "null",
|
|
})
|
|
continue
|
|
}
|
|
deleteVersionId = "null"
|
|
// Note: For suspended versioning, we don't set isDeleteMarker=true
|
|
// because we actually deleted the object, not created a delete marker
|
|
}
|
|
}
|
|
|
|
// Add to successful deletions with version info
|
|
deletedObject := ObjectIdentifier{
|
|
Key: object.Key,
|
|
VersionId: deleteVersionId,
|
|
DeleteMarker: isDeleteMarker,
|
|
}
|
|
|
|
// For delete markers, also set DeleteMarkerVersionId field
|
|
if isDeleteMarker {
|
|
deletedObject.DeleteMarkerVersionId = deleteVersionId
|
|
// Don't set VersionId for delete markers, use DeleteMarkerVersionId instead
|
|
deletedObject.VersionId = ""
|
|
}
|
|
if !deleteObjects.Quiet {
|
|
deletedObjects = append(deletedObjects, deletedObject)
|
|
}
|
|
if isDeleteMarker {
|
|
// For delete markers, we don't need to track directories for cleanup
|
|
continue
|
|
}
|
|
} else {
|
|
// Handle non-versioned delete (original logic)
|
|
lastSeparator := strings.LastIndex(object.Key, "/")
|
|
parentDirectoryPath, entryName, isDeleteData, isRecursive := "", object.Key, true, false
|
|
if lastSeparator > 0 && lastSeparator+1 < len(object.Key) {
|
|
entryName = object.Key[lastSeparator+1:]
|
|
parentDirectoryPath = "/" + object.Key[:lastSeparator]
|
|
}
|
|
parentDirectoryPath = fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, parentDirectoryPath)
|
|
|
|
err := doDeleteEntry(client, parentDirectoryPath, entryName, isDeleteData, isRecursive)
|
|
if err == nil {
|
|
deletedObjects = append(deletedObjects, object)
|
|
} else if strings.Contains(err.Error(), filer.MsgFailDelNonEmptyFolder) {
|
|
deletedObjects = append(deletedObjects, object)
|
|
} else {
|
|
deleteErrors = append(deleteErrors, DeleteError{
|
|
Code: "",
|
|
Message: err.Error(),
|
|
Key: object.Key,
|
|
VersionId: object.VersionId,
|
|
})
|
|
}
|
|
}
|
|
|
|
if auditLog != nil {
|
|
auditLog.Key = object.Key
|
|
s3err.PostAccessLog(*auditLog)
|
|
}
|
|
}
|
|
|
|
// Note: Empty folder cleanup is now handled asynchronously by EmptyFolderCleaner
|
|
// which listens to metadata events and uses consistent hashing for coordination
|
|
|
|
return nil
|
|
})
|
|
|
|
deleteResp := DeleteObjectsResponse{}
|
|
if !deleteObjects.Quiet {
|
|
deleteResp.DeletedObjects = deletedObjects
|
|
}
|
|
deleteResp.Errors = deleteErrors
|
|
stats_collect.RecordBucketActiveTime(bucket)
|
|
stats_collect.S3DeletedObjectsCounter.WithLabelValues(bucket).Add(float64(len(deletedObjects)))
|
|
|
|
writeSuccessResponseXML(w, r, deleteResp)
|
|
|
|
}
|