mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-29 11:15:34 +00:00
* s3: ignore empty intermediate directories in bucketHasUserObjects (#11490) * s3: keep nested reserved-named dirs from hiding user objects Reserved folders (.uploads, *.versions) are internal only at the bucket root; deeper entries with those names are user key prefixes and must be walked. Also treat a missing subdirectory as empty via isFilerNotFound (list errors cross gRPC as status errors, not the sentinel), let names containing backslashes count as objects, and walk iteratively so empty chains deeper than the old scan depth no longer report non-empty. * s3: treat reserved-named directories as internal at every level Object listing interprets .uploads and *.versions directories as internal storage wherever they appear, so walking them during the emptiness check would report invisible version remnants as user objects and block deletion. A reserved name on a file still counts, matching listing which only special-cases directories. * s3: count explicit directory objects under reserved names A directory object created by PutObject (MIME or prefix-object marker set) is user data even when named .uploads or *.versions; only a plain directory with a reserved name is internal storage. --------- Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
This commit is contained in:
yi111
Chris Lu
@@ -0,0 +1,343 @@
|
||||
package s3api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
type fakeBucketDeleteFiler struct {
|
||||
filer_pb.UnimplementedSeaweedFilerServer
|
||||
entriesByDir map[string][]*filer_pb.Entry
|
||||
deleteReq *filer_pb.DeleteEntryRequest
|
||||
}
|
||||
|
||||
func (f *fakeBucketDeleteFiler) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
|
||||
entries, ok := f.entriesByDir[req.Directory]
|
||||
if !ok {
|
||||
return nil, filer_pb.ErrNotFound
|
||||
}
|
||||
for _, e := range entries {
|
||||
if e.Name == req.Name {
|
||||
return &filer_pb.LookupDirectoryEntryResponse{Entry: e}, nil
|
||||
}
|
||||
}
|
||||
return nil, filer_pb.ErrNotFound
|
||||
}
|
||||
|
||||
func (f *fakeBucketDeleteFiler) ListEntries(req *filer_pb.ListEntriesRequest, stream filer_pb.SeaweedFiler_ListEntriesServer) error {
|
||||
entries := f.entriesByDir[req.Directory]
|
||||
if inPrefix := req.Prefix; inPrefix != "" && inPrefix != "/" {
|
||||
filtered := make([]*filer_pb.Entry, 0)
|
||||
for _, e := range entries {
|
||||
if strings.HasPrefix(e.Name, inPrefix) {
|
||||
filtered = append(filtered, e)
|
||||
}
|
||||
}
|
||||
entries = filtered
|
||||
}
|
||||
if req.StartFromFileName != "" {
|
||||
filtered := make([]*filer_pb.Entry, 0)
|
||||
for _, e := range entries {
|
||||
if e.Name > req.StartFromFileName || (req.InclusiveStartFrom && e.Name == req.StartFromFileName) {
|
||||
filtered = append(filtered, e)
|
||||
}
|
||||
}
|
||||
entries = filtered
|
||||
}
|
||||
if req.Limit > 0 && int(req.Limit) < len(entries) {
|
||||
entries = entries[:req.Limit]
|
||||
}
|
||||
for _, entry := range entries {
|
||||
if err := stream.Send(&filer_pb.ListEntriesResponse{Entry: entry}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeBucketDeleteFiler) DeleteEntry(ctx context.Context, req *filer_pb.DeleteEntryRequest) (*filer_pb.DeleteEntryResponse, error) {
|
||||
f.deleteReq = req
|
||||
return &filer_pb.DeleteEntryResponse{}, nil
|
||||
}
|
||||
|
||||
func newBucketDeleteTestServer(t *testing.T, f *fakeBucketDeleteFiler, allowDeleteBucketNotEmpty bool) *S3ApiServer {
|
||||
t.Helper()
|
||||
s3a := newFailoverTestServer(t, startFakeFiler(t, f))
|
||||
s3a.option.BucketsPath = "/buckets"
|
||||
s3a.option.AllowDeleteBucketNotEmpty = allowDeleteBucketNotEmpty
|
||||
s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
|
||||
s3a.iam = &IdentityAccessManagement{}
|
||||
return s3a
|
||||
}
|
||||
|
||||
func TestBucketHasUserObjects_EmptyDirectories(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
entriesByDir map[string][]*filer_pb.Entry
|
||||
wantHasUser bool
|
||||
}{
|
||||
{
|
||||
name: "completely empty bucket",
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets/b": {},
|
||||
},
|
||||
wantHasUser: false,
|
||||
},
|
||||
{
|
||||
name: "bucket with root file",
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets/b": {
|
||||
{Name: "file.txt", IsDirectory: false},
|
||||
},
|
||||
},
|
||||
wantHasUser: true,
|
||||
},
|
||||
{
|
||||
name: "bucket with leftover empty directory from rm --recursive",
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets/b": {
|
||||
{Name: "data", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mime: ""}},
|
||||
},
|
||||
"/buckets/b/data": {},
|
||||
},
|
||||
wantHasUser: false,
|
||||
},
|
||||
{
|
||||
name: "bucket with nested leftover empty directories",
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets/b": {
|
||||
{Name: "data", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mime: ""}},
|
||||
},
|
||||
"/buckets/b/data": {
|
||||
{Name: "2026", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mime: ""}},
|
||||
},
|
||||
"/buckets/b/data/2026": {},
|
||||
},
|
||||
wantHasUser: false,
|
||||
},
|
||||
{
|
||||
name: "bucket with nested directory containing a file",
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets/b": {
|
||||
{Name: "data", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mime: ""}},
|
||||
},
|
||||
"/buckets/b/data": {
|
||||
{Name: "2026", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mime: ""}},
|
||||
},
|
||||
"/buckets/b/data/2026": {
|
||||
{Name: "report.pdf", IsDirectory: false},
|
||||
},
|
||||
},
|
||||
wantHasUser: true,
|
||||
},
|
||||
{
|
||||
name: "bucket with explicit directory object (MIME set)",
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets/b": {
|
||||
{Name: "logs", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mime: "application/x-directory"}},
|
||||
},
|
||||
"/buckets/b/logs": {},
|
||||
},
|
||||
wantHasUser: true,
|
||||
},
|
||||
{
|
||||
name: "bucket with only reserved folders (.uploads, *.versions)",
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets/b": {
|
||||
{Name: s3_constants.MultipartUploadsFolder, IsDirectory: true},
|
||||
{Name: "oldfile.txt" + s3_constants.VersionsFolder, IsDirectory: true},
|
||||
},
|
||||
},
|
||||
wantHasUser: false,
|
||||
},
|
||||
{
|
||||
name: "nested versions-named directories are internal",
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets/b": {
|
||||
{Name: "logs", IsDirectory: true},
|
||||
},
|
||||
"/buckets/b/logs": {
|
||||
{Name: "foo" + s3_constants.VersionsFolder, IsDirectory: true},
|
||||
},
|
||||
"/buckets/b/logs/foo" + s3_constants.VersionsFolder: {
|
||||
{Name: "v1", IsDirectory: false},
|
||||
},
|
||||
},
|
||||
wantHasUser: false,
|
||||
},
|
||||
{
|
||||
name: "nested uploads-named directories are internal",
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets/b": {
|
||||
{Name: "data", IsDirectory: true},
|
||||
},
|
||||
"/buckets/b/data": {
|
||||
{Name: s3_constants.MultipartUploadsFolder, IsDirectory: true},
|
||||
},
|
||||
"/buckets/b/data/" + s3_constants.MultipartUploadsFolder: {
|
||||
{Name: "part-1", IsDirectory: false},
|
||||
},
|
||||
},
|
||||
wantHasUser: false,
|
||||
},
|
||||
{
|
||||
name: "directory object with a reserved name counts",
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets/b": {
|
||||
{Name: "data", IsDirectory: true},
|
||||
},
|
||||
"/buckets/b/data": {
|
||||
{Name: s3_constants.MultipartUploadsFolder, IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mime: s3_constants.FolderMimeType}},
|
||||
},
|
||||
},
|
||||
wantHasUser: true,
|
||||
},
|
||||
{
|
||||
name: "file with a reserved name counts",
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets/b": {
|
||||
{Name: "dir", IsDirectory: true},
|
||||
},
|
||||
"/buckets/b/dir": {
|
||||
{Name: "data" + s3_constants.VersionsFolder, IsDirectory: false},
|
||||
},
|
||||
},
|
||||
wantHasUser: true,
|
||||
},
|
||||
{
|
||||
name: "file with backslash in name",
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets/b": {
|
||||
{Name: `a\b`, IsDirectory: false},
|
||||
},
|
||||
},
|
||||
wantHasUser: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
f := &fakeBucketDeleteFiler{entriesByDir: tc.entriesByDir}
|
||||
s3a := newBucketDeleteTestServer(t, f, false)
|
||||
|
||||
got, err := s3a.bucketHasUserObjects("b")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, tc.wantHasUser, got)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestBucketHasUserObjects_DeepEmptyChain(t *testing.T) {
|
||||
entriesByDir := map[string][]*filer_pb.Entry{
|
||||
"/buckets/b": {{Name: "d0", IsDirectory: true}},
|
||||
}
|
||||
dir := "/buckets/b/d0"
|
||||
for i := 1; i < 500; i++ {
|
||||
child := fmt.Sprintf("d%d", i)
|
||||
entriesByDir[dir] = []*filer_pb.Entry{{Name: child, IsDirectory: true}}
|
||||
dir = dir + "/" + child
|
||||
}
|
||||
entriesByDir[dir] = []*filer_pb.Entry{}
|
||||
|
||||
f := &fakeBucketDeleteFiler{entriesByDir: entriesByDir}
|
||||
s3a := newBucketDeleteTestServer(t, f, false)
|
||||
|
||||
got, err := s3a.bucketHasUserObjects("b")
|
||||
require.NoError(t, err)
|
||||
assert.False(t, got)
|
||||
}
|
||||
|
||||
func TestDeleteBucketHandler_EmptyDirectoriesAllowedWhenNotEmptyFalse(t *testing.T) {
|
||||
// A bucket that holds only an empty directory leftover from deleted objects
|
||||
// must succeed when AllowDeleteBucketNotEmpty is false.
|
||||
f := &fakeBucketDeleteFiler{
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets": {
|
||||
{Name: "repro-bucket", IsDirectory: true},
|
||||
},
|
||||
"/buckets/repro-bucket": {
|
||||
{Name: "data", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mime: ""}},
|
||||
},
|
||||
"/buckets/repro-bucket/data": {},
|
||||
},
|
||||
}
|
||||
s3a := newBucketDeleteTestServer(t, f, false)
|
||||
s3a.bucketConfigCache.Set("repro-bucket", &BucketConfig{Name: "repro-bucket"})
|
||||
|
||||
req := httptest.NewRequest(http.MethodDelete, "/repro-bucket", nil)
|
||||
req = mux.SetURLVars(req, map[string]string{"bucket": "repro-bucket"})
|
||||
rr := httptest.NewRecorder()
|
||||
|
||||
s3a.DeleteBucketHandler(rr, req)
|
||||
|
||||
assert.Equal(t, http.StatusNoContent, rr.Code, "deleting bucket with only empty directory must succeed: %s", rr.Body.String())
|
||||
assert.NotNil(t, f.deleteReq)
|
||||
assert.Equal(t, "/buckets", f.deleteReq.Directory)
|
||||
assert.Equal(t, "repro-bucket", f.deleteReq.Name)
|
||||
assert.True(t, f.deleteReq.IsRecursive)
|
||||
}
|
||||
|
||||
func TestDeleteBucketHandler_RefusesBucketWithFilesWhenNotEmptyFalse(t *testing.T) {
|
||||
f := &fakeBucketDeleteFiler{
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets": {
|
||||
{Name: "non-empty-bucket", IsDirectory: true},
|
||||
},
|
||||
"/buckets/non-empty-bucket": {
|
||||
{Name: "data", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mime: ""}},
|
||||
},
|
||||
"/buckets/non-empty-bucket/data": {
|
||||
{Name: "hello.txt", IsDirectory: false},
|
||||
},
|
||||
},
|
||||
}
|
||||
s3a := newBucketDeleteTestServer(t, f, false)
|
||||
s3a.bucketConfigCache.Set("non-empty-bucket", &BucketConfig{Name: "non-empty-bucket"})
|
||||
|
||||
req := httptest.NewRequest(http.MethodDelete, "/non-empty-bucket", nil)
|
||||
req = mux.SetURLVars(req, map[string]string{"bucket": "non-empty-bucket"})
|
||||
rr := httptest.NewRecorder()
|
||||
|
||||
s3a.DeleteBucketHandler(rr, req)
|
||||
|
||||
assert.Equal(t, http.StatusConflict, rr.Code)
|
||||
assert.Contains(t, rr.Body.String(), "<Code>BucketNotEmpty</Code>")
|
||||
assert.Nil(t, f.deleteReq, "delete request must not have been sent to filer")
|
||||
}
|
||||
|
||||
func TestDeleteBucketHandler_RefusesBucketWithDirectoryObjectWhenNotEmptyFalse(t *testing.T) {
|
||||
f := &fakeBucketDeleteFiler{
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets": {
|
||||
{Name: "dir-obj-bucket", IsDirectory: true},
|
||||
},
|
||||
"/buckets/dir-obj-bucket": {
|
||||
{Name: "photos", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mime: "application/x-directory"}},
|
||||
},
|
||||
"/buckets/dir-obj-bucket/photos": {},
|
||||
},
|
||||
}
|
||||
s3a := newBucketDeleteTestServer(t, f, false)
|
||||
s3a.bucketConfigCache.Set("dir-obj-bucket", &BucketConfig{Name: "dir-obj-bucket"})
|
||||
|
||||
req := httptest.NewRequest(http.MethodDelete, "/dir-obj-bucket", nil)
|
||||
req = mux.SetURLVars(req, map[string]string{"bucket": "dir-obj-bucket"})
|
||||
rr := httptest.NewRecorder()
|
||||
|
||||
s3a.DeleteBucketHandler(rr, req)
|
||||
|
||||
assert.Equal(t, http.StatusConflict, rr.Code)
|
||||
assert.Contains(t, rr.Body.String(), "<Code>BucketNotEmpty</Code>")
|
||||
assert.Nil(t, f.deleteReq, "delete request must not have been sent to filer")
|
||||
}
|
||||
@@ -503,31 +503,56 @@ func (s3a *S3ApiServer) DeleteBucketHandler(w http.ResponseWriter, r *http.Reque
|
||||
s3err.WriteEmptyResponse(w, r, http.StatusNoContent)
|
||||
}
|
||||
|
||||
// bucketHasUserObjects checks whether a bucket contains any non-special entries.
|
||||
// Special entries (.uploads, *.versions) are internal to S3 and don't count as user objects.
|
||||
// bucketHasUserObjects checks whether a bucket contains any user objects.
|
||||
// Empty directories left behind by deleted objects and internal folders
|
||||
// (.uploads, *.versions) do not count as user objects.
|
||||
func (s3a *S3ApiServer) bucketHasUserObjects(bucket string) (bool, error) {
|
||||
bucketPath := s3a.option.BucketsPath + "/" + bucket
|
||||
startFrom := ""
|
||||
// Start with a small batch — most non-empty buckets have a real object early.
|
||||
// If we only find special entries, switch to larger batches to page through quickly.
|
||||
limit := uint32(10)
|
||||
for {
|
||||
entries, isLast, err := s3a.list(bucketPath, "", startFrom, false, limit)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
for _, entry := range entries {
|
||||
if entry.Name != s3_constants.MultipartUploadsFolder &&
|
||||
!strings.HasSuffix(entry.Name, s3_constants.VersionsFolder) {
|
||||
return true, nil
|
||||
return s3a.dirHasUserObjects(s3a.option.BucketsPath + "/" + bucket)
|
||||
}
|
||||
|
||||
func (s3a *S3ApiServer) dirHasUserObjects(root string) (bool, error) {
|
||||
dirs := []string{root}
|
||||
for len(dirs) > 0 {
|
||||
dir := dirs[len(dirs)-1]
|
||||
dirs = dirs[:len(dirs)-1]
|
||||
startFrom := ""
|
||||
// Start with a small batch — most non-empty buckets have a real object
|
||||
// early; switch to larger batches to page through quickly.
|
||||
limit := uint32(10)
|
||||
for {
|
||||
entries, isLast, err := s3a.list(dir, "", startFrom, false, limit)
|
||||
if err != nil {
|
||||
if isFilerNotFound(err) {
|
||||
break // the directory was deleted between listing and walking it
|
||||
}
|
||||
return false, err
|
||||
}
|
||||
startFrom = entry.Name
|
||||
for _, entry := range entries {
|
||||
startFrom = entry.Name
|
||||
if entry.Name == "" || entry.Name == "." || entry.Name == ".." || strings.Contains(entry.Name, "/") {
|
||||
continue
|
||||
}
|
||||
if !entry.IsDirectory {
|
||||
return true, nil
|
||||
}
|
||||
// An explicit directory object counts even under a reserved name.
|
||||
if entry.IsDirectoryKeyObject() {
|
||||
return true, nil
|
||||
}
|
||||
// Internal folders are skipped by object listing at every level,
|
||||
// so a directory with a reserved name never counts.
|
||||
if isReservedDirectoryName(entry.Name) {
|
||||
continue
|
||||
}
|
||||
dirs = append(dirs, dir+"/"+entry.Name)
|
||||
}
|
||||
if isLast || len(entries) == 0 {
|
||||
break
|
||||
}
|
||||
limit = 1000
|
||||
}
|
||||
if isLast {
|
||||
return false, nil
|
||||
}
|
||||
limit = 1000
|
||||
}
|
||||
return false, nil
|
||||
}
|
||||
|
||||
// hasObjectsWithActiveLocks checks if any objects in the bucket have active retention or legal hold
|
||||
|
||||
Reference in New Issue
Block a user