From bdc37a1e868d4963cff2a2d76bdcd66c78e804a9 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 17 Sep 2026 20:30:53 -0700 Subject: [PATCH] mount/shell: bucket allow-empty-folders toggle, mount keeps explicit false (#11370) * mount: keep a deliberate bucket allow-empty-folders setting * shell: s3.bucket.allowEmptyFolders toggles the empty folder cleaner * shell: guard allow-empty-folders toggle with expected extended attrs * filer: drop cached empty-folder policy on bucket entry update * mount: guard allow-empty-folders write with expected extended attrs * filer: skip caching a stale cleanup policy read across an update * mount, shell: snapshot the full extended attributes for update preconditions * filer: fail closed and invalidate on all bucket entry events for cleanup policy * filer: key the cleanup policy generation by bucket * filer: requeue cleanup when the bucket policy cannot be loaded * filer: expire idle cleanup policy generations * filer: skip requeueing cleanup after the cleaner stops * filer: bound cleanup retries on repeated policy failures * filer: cover cleanup requeue on repeated policy failures * filer: keep cleanup policy generations while reads are in flight * filer: exercise the cleanup queue lifecycle in the retry-cap test --- weed/command/mount_common.go | 62 +++++--- weed/command/mount_common_test.go | 113 ++++++++++++- .../empty_folder_cleaner.go | 142 +++++++++++++---- .../empty_folder_cleaner_test.go | 150 ++++++++++++++++++ weed/filer/filer_on_meta_event.go | 9 ++ weed/pb/filer_pb/filer_pb_helper.go | 16 ++ .../command_s3_bucket_allow_empty_folders.go | 130 +++++++++++++++ 7 files changed, 569 insertions(+), 53 deletions(-) create mode 100644 weed/shell/command_s3_bucket_allow_empty_folders.go diff --git a/weed/command/mount_common.go b/weed/command/mount_common.go index 18d92b58a..bbabe56ac 100644 --- a/weed/command/mount_common.go +++ b/weed/command/mount_common.go @@ -17,7 +17,9 @@ import ( "time" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/reflection" + "google.golang.org/grpc/status" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/mount" @@ -67,36 +69,46 @@ func ensureBucketAllowEmptyFolders(ctx context.Context, filerClient filer_pb.Fil return nil } - entry, _, _, err := filer_pb.GetEntry(ctx, filerClient, util.FullPath(bucketPath)) - if err != nil { - return err - } - if entry == nil { - return fmt.Errorf("bucket %s not found", bucketPath) - } - - if entry.Extended == nil { - entry.Extended = make(map[string][]byte) - } - if strings.EqualFold(strings.TrimSpace(string(entry.Extended[s3_constants.ExtAllowEmptyFolders])), "true") { - return nil - } - - entry.Extended[s3_constants.ExtAllowEmptyFolders] = []byte("true") - bucketFullPath := util.FullPath(bucketPath) parent, _ := bucketFullPath.DirAndName() - if err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return filer_pb.UpdateEntry(ctx, client, &filer_pb.UpdateEntryRequest{ - Directory: parent, - Entry: entry, + + for attempt := 0; attempt < 5; attempt++ { + entry, _, _, err := filer_pb.GetEntry(ctx, filerClient, util.FullPath(bucketPath)) + if err != nil { + return err + } + if entry == nil { + return fmt.Errorf("bucket %s not found", bucketPath) + } + + if value := strings.TrimSpace(string(entry.Extended[s3_constants.ExtAllowEmptyFolders])); value != "" { + return nil + } + + expected := filer_pb.SnapshotExtended(entry.Extended, s3_constants.ExtAllowEmptyFolders) + if entry.Extended == nil { + entry.Extended = make(map[string][]byte) + } + entry.Extended[s3_constants.ExtAllowEmptyFolders] = []byte("true") + + err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + _, err := client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{ + Directory: parent, + Entry: entry, + ExpectedExtended: expected, + }) + return err }) - }); err != nil { - return err + if err == nil { + glog.V(3).Infof("RunMount: set bucket %s %s=true", bucketPath, s3_constants.ExtAllowEmptyFolders) + return nil + } + if status.Code(err) != codes.FailedPrecondition { + return err + } } - glog.V(3).Infof("RunMount: set bucket %s %s=true", bucketPath, s3_constants.ExtAllowEmptyFolders) - return nil + return fmt.Errorf("bucket %s attributes keep changing", bucketPath) } func bucketPathForMountRoot(mountRoot, bucketRootPath string) (string, bool) { diff --git a/weed/command/mount_common_test.go b/weed/command/mount_common_test.go index 17f40a954..48878cca5 100644 --- a/weed/command/mount_common_test.go +++ b/weed/command/mount_common_test.go @@ -2,7 +2,18 @@ package command -import "testing" +import ( + "bytes" + "context" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" +) func Test_volumeName(t *testing.T) { tests := []struct { @@ -102,3 +113,103 @@ func Test_volumeName(t *testing.T) { }) } } + +type allowEmptyFoldersInnerClient struct { + filer_pb.SeaweedFilerClient + entry *filer_pb.Entry + beforeUpdate func() + lookups int + updateAttempts int + updates []*filer_pb.Entry +} + +func (c *allowEmptyFoldersInnerClient) LookupDirectoryEntry(_ context.Context, _ *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { + c.lookups++ + return &filer_pb.LookupDirectoryEntryResponse{Entry: proto.Clone(c.entry).(*filer_pb.Entry)}, nil +} + +func (c *allowEmptyFoldersInnerClient) UpdateEntry(_ context.Context, in *filer_pb.UpdateEntryRequest, _ ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + c.updateAttempts++ + if c.beforeUpdate != nil { + c.beforeUpdate() + } + for key, expected := range in.ExpectedExtended { + actual, found := c.entry.Extended[key] + if found && !bytes.Equal(actual, expected) || !found && len(expected) > 0 { + return nil, status.Error(codes.FailedPrecondition, "extended attribute changed") + } + } + c.entry = proto.Clone(in.Entry).(*filer_pb.Entry) + c.updates = append(c.updates, in.Entry) + return &filer_pb.UpdateEntryResponse{}, nil +} + +type allowEmptyFoldersFilerClient struct { + inner *allowEmptyFoldersInnerClient +} + +func (c *allowEmptyFoldersFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error { + return fn(c.inner) +} +func (c *allowEmptyFoldersFilerClient) AdjustedUrl(_ *filer_pb.Location) string { return "" } +func (c *allowEmptyFoldersFilerClient) GetDataCenter() string { return "" } + +func Test_ensureBucketAllowEmptyFolders(t *testing.T) { + bucketEntry := func(attrValue string) *filer_pb.Entry { + entry := &filer_pb.Entry{Name: "b1", IsDirectory: true, Extended: map[string][]byte{}} + if attrValue != "" { + entry.Extended[s3_constants.ExtAllowEmptyFolders] = []byte(attrValue) + } + return entry + } + + tests := []struct { + name string + mountRoot string + attrValue string + beforeUpdate func(inner *allowEmptyFoldersInnerClient) + wantLookups int + wantUpdates int + wantAttrValue string + }{ + {name: "no policy allows empty folders", mountRoot: "/buckets/b1", wantLookups: 1, wantUpdates: 1, wantAttrValue: "true"}, + {name: "explicit true is kept", mountRoot: "/buckets/b1", attrValue: "true", wantLookups: 1, wantUpdates: 0}, + {name: "explicit false is kept", mountRoot: "/buckets/b1", attrValue: "false", wantLookups: 1, wantUpdates: 0}, + {name: "subdirectory mount leaves the bucket alone", mountRoot: "/buckets/b1/sub", wantLookups: 0, wantUpdates: 0}, + { + name: "a concurrent false wins over the mount default", + mountRoot: "/buckets/b1", + beforeUpdate: func(inner *allowEmptyFoldersInnerClient) { + inner.entry.Extended[s3_constants.ExtAllowEmptyFolders] = []byte("false") + }, + wantLookups: 2, + wantUpdates: 0, + wantAttrValue: "false", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + inner := &allowEmptyFoldersInnerClient{entry: bucketEntry(tt.attrValue)} + if tt.beforeUpdate != nil { + inner.beforeUpdate = func() { tt.beforeUpdate(inner) } + } + filerClient := &allowEmptyFoldersFilerClient{inner: inner} + + if err := ensureBucketAllowEmptyFolders(context.Background(), filerClient, tt.mountRoot, "/buckets"); err != nil { + t.Fatalf("ensureBucketAllowEmptyFolders: %v", err) + } + if inner.lookups != tt.wantLookups { + t.Errorf("lookups = %d, want %d", inner.lookups, tt.wantLookups) + } + if len(inner.updates) != tt.wantUpdates { + t.Fatalf("updates = %d, want %d", len(inner.updates), tt.wantUpdates) + } + if tt.wantAttrValue != "" { + got := string(inner.entry.Extended[s3_constants.ExtAllowEmptyFolders]) + if got != tt.wantAttrValue { + t.Errorf("%s = %q, want %q", s3_constants.ExtAllowEmptyFolders, got, tt.wantAttrValue) + } + } + }) + } +} diff --git a/weed/filer/empty_folder_cleanup/empty_folder_cleaner.go b/weed/filer/empty_folder_cleanup/empty_folder_cleaner.go index d7acb84d0..b85b94d20 100644 --- a/weed/filer/empty_folder_cleanup/empty_folder_cleaner.go +++ b/weed/filer/empty_folder_cleanup/empty_folder_cleaner.go @@ -2,6 +2,7 @@ package empty_folder_cleanup import ( "context" + "fmt" "os" "sort" "strings" @@ -17,12 +18,13 @@ import ( ) const ( - DefaultMaxCountCheck = 1000 - DefaultCacheExpiry = 5 * time.Minute - DefaultQueueMaxSize = 1000 - DefaultQueueMaxAge = 2 * time.Minute - DefaultProcessorSleep = 30 * time.Second // How often to check queue - DefaultMaxDeletedKept = 10000 // Deleted folders remembered for the restore check + DefaultMaxCountCheck = 1000 + DefaultCacheExpiry = 5 * time.Minute + DefaultQueueMaxSize = 1000 + DefaultQueueMaxAge = 2 * time.Minute + DefaultProcessorSleep = 30 * time.Second // How often to check queue + DefaultMaxDeletedKept = 10000 // Deleted folders remembered for the restore check + DefaultMaxPolicyFailures = 3 // Consecutive policy-load failures before a folder is dropped // How long a deleted folder is kept so that a create event arriving for it can // still put it back. It bounds how far behind the event stream may run, not how // long the race window is. @@ -59,10 +61,17 @@ type FilerOperations interface { // folderState tracks the state of a folder for empty folder cleanup type folderState struct { - roughCount int // Cached rough count (up to maxCountCheck) - lastAddTime time.Time // Last time an item was added - lastDelTime time.Time // Last time an item was deleted - lastCheck time.Time // Last time we checked the actual count + roughCount int // Cached rough count (up to maxCountCheck) + lastAddTime time.Time // Last time an item was added + lastDelTime time.Time // Last time an item was deleted + lastCheck time.Time // Last time we checked the actual count + policyFailures int // Consecutive bucket policy load failures +} + +type bucketPolicyGen struct { + gen uint64 + lastBump time.Time + inflight int } type bucketCleanupPolicyState struct { @@ -82,6 +91,7 @@ type EmptyFolderCleaner struct { mu sync.RWMutex folderCounts map[string]*folderState // Rough count cache bucketCleanupPolicies map[string]*bucketCleanupPolicyState // bucket path -> cleanup policy cache + policyGen map[string]*bucketPolicyGen // bucket path -> generation, bumped on bucket entry updates to invalidate in-flight policy loads // Folders deleted recently, kept so that a create event arriving for one of them // can put it back @@ -115,6 +125,7 @@ func NewEmptyFolderCleaner(filer FilerOperations, lockRing *lock_manager.LockRin host: host, folderCounts: make(map[string]*folderState), bucketCleanupPolicies: make(map[string]*bucketCleanupPolicyState), + policyGen: make(map[string]*bucketPolicyGen), deleted: make(map[string]*deletedFolder), cleanupQueue: NewCleanupQueue(DefaultQueueMaxSize, cleanupDelay), maxCountCheck: DefaultMaxCountCheck, @@ -251,6 +262,33 @@ func (efc *EmptyFolderCleaner) OnCreateEvent(directory string, entryName string, } } +// InvalidateBucketPolicy drops the cached cleanup policy when a bucket entry changes +func (efc *EmptyFolderCleaner) InvalidateBucketPolicy(directory string, entryName string, isDirectory bool) { + if !isDirectory || directory != efc.bucketPath { + return + } + + efc.mu.Lock() + defer efc.mu.Unlock() + + if !efc.enabled { + return + } + + if efc.policyGen == nil { + efc.policyGen = make(map[string]*bucketPolicyGen) + } + path := string(util.NewFullPath(directory, entryName)) + gen := efc.policyGen[path] + if gen == nil { + gen = &bucketPolicyGen{} + efc.policyGen[path] = gen + } + gen.gen++ + gen.lastBump = time.Now() + delete(efc.bucketCleanupPolicies, path) +} + // cleanupProcessor runs in background and processes the cleanup queue func (efc *EmptyFolderCleaner) cleanupProcessor() { ticker := time.NewTicker(efc.processorSleep) @@ -491,9 +529,28 @@ func (efc *EmptyFolderCleaner) executeCleanup(folder string, triggeredBy string) return } glog.V(2).Infof("EmptyFolderCleaner: failed to load bucket cleanup policy for folder %s (triggered by %s): %v", folder, triggeredBy, err) + efc.mu.Lock() + if efc.enabled { + state, exists := efc.folderCounts[folder] + if !exists { + state = &folderState{} + efc.folderCounts[folder] = state + } + state.policyFailures++ + if state.policyFailures <= DefaultMaxPolicyFailures { + efc.cleanupQueue.Add(folder, triggeredBy, time.Now()) + } + } + efc.mu.Unlock() return } + efc.mu.Lock() + if state, exists := efc.folderCounts[folder]; exists { + state.policyFailures = 0 + } + efc.mu.Unlock() + if !autoRemove { glog.V(3).Infof("EmptyFolderCleaner: skipping folder %s (triggered by %s), bucket %s auto-remove-empty-folders disabled (source=%s attr=%s)", folder, triggeredBy, bucketPath, source, attrValue) @@ -617,25 +674,47 @@ func (efc *EmptyFolderCleaner) getBucketCleanupPolicy(ctx context.Context, folde } efc.mu.RUnlock() - attrs, err := efc.filer.GetEntryAttributes(ctx, util.FullPath(bucketPath)) - if err != nil { - return "", true, "", "", err + for attempt := 0; attempt < 3; attempt++ { + efc.mu.Lock() + if efc.policyGen == nil { + efc.policyGen = make(map[string]*bucketPolicyGen) + } + state := efc.policyGen[bucketPath] + if state == nil { + state = &bucketPolicyGen{} + efc.policyGen[bucketPath] = state + } + state.inflight++ + gen := state.gen + efc.mu.Unlock() + + attrs, err := efc.filer.GetEntryAttributes(ctx, util.FullPath(bucketPath)) + + efc.mu.Lock() + state.inflight-- + if err != nil { + efc.mu.Unlock() + return "", true, "", "", err + } + autoRemove, attrValue = autoRemoveEmptyFoldersEnabled(attrs) + if gen != state.gen { + efc.mu.Unlock() + continue + } + if efc.bucketCleanupPolicies == nil { + efc.bucketCleanupPolicies = make(map[string]*bucketCleanupPolicyState) + } + efc.bucketCleanupPolicies[bucketPath] = &bucketCleanupPolicyState{ + autoRemove: autoRemove, + attrValue: attrValue, + lastCheck: now, + } + efc.mu.Unlock() + + return bucketPath, autoRemove, "filer", attrValue, nil } - autoRemove, attrValue = autoRemoveEmptyFoldersEnabled(attrs) - - efc.mu.Lock() - if efc.bucketCleanupPolicies == nil { - efc.bucketCleanupPolicies = make(map[string]*bucketCleanupPolicyState) - } - efc.bucketCleanupPolicies[bucketPath] = &bucketCleanupPolicyState{ - autoRemove: autoRemove, - attrValue: attrValue, - lastCheck: now, - } - efc.mu.Unlock() - - return bucketPath, autoRemove, "filer", attrValue, nil + return "", false, "", "", fmt.Errorf("bucket cleanup policy changed during read: %s", bucketPath) } // isCatalogEntry reports whether the directory is an s3tables catalog record. @@ -775,6 +854,14 @@ func (efc *EmptyFolderCleaner) evictStaleCacheEntries() { } } + // A generation stays while a read that captured it can still return; an idle + // one drops after the expiry so bucket churn cannot grow the map forever. + for bucketPath, gen := range efc.policyGen { + if gen.inflight == 0 && now.Sub(gen.lastBump) > efc.cacheExpiry { + delete(efc.policyGen, bucketPath) + } + } + if expiredCount > 0 { glog.V(3).Infof("EmptyFolderCleaner: evicted %d stale cache entries", expiredCount) } @@ -791,6 +878,7 @@ func (efc *EmptyFolderCleaner) Stop() { efc.cleanupQueue.Clear() efc.folderCounts = make(map[string]*folderState) // Clear cache on stop efc.bucketCleanupPolicies = make(map[string]*bucketCleanupPolicyState) + efc.policyGen = make(map[string]*bucketPolicyGen) efc.deleted, efc.deletedDropped = make(map[string]*deletedFolder), 0 } diff --git a/weed/filer/empty_folder_cleanup/empty_folder_cleaner_test.go b/weed/filer/empty_folder_cleanup/empty_folder_cleaner_test.go index e163aba13..bf673b630 100644 --- a/weed/filer/empty_folder_cleanup/empty_folder_cleaner_test.go +++ b/weed/filer/empty_folder_cleanup/empty_folder_cleaner_test.go @@ -1074,6 +1074,156 @@ func TestEmptyFolderCleaner_executeCleanup_bucketPolicyDisabledSkips(t *testing. } } +func TestEmptyFolderCleaner_InvalidateBucketPolicy(t *testing.T) { + cleaner := &EmptyFolderCleaner{ + bucketPath: "/buckets", + enabled: true, + bucketCleanupPolicies: map[string]*bucketCleanupPolicyState{ + "/buckets/test": {}, + }, + } + + cleaner.InvalidateBucketPolicy("/buckets", "test", true) + if _, found := cleaner.bucketCleanupPolicies["/buckets/test"]; found { + t.Fatal("expected cached bucket policy to be evicted") + } + + cleaner.bucketCleanupPolicies["/buckets/test"] = &bucketCleanupPolicyState{} + cleaner.InvalidateBucketPolicy("/buckets/test", "dir", true) + cleaner.InvalidateBucketPolicy("/buckets", "test", false) + if _, found := cleaner.bucketCleanupPolicies["/buckets/test"]; !found { + t.Fatal("expected unrelated updates to keep the cached policy") + } +} + +func TestEmptyFolderCleaner_getBucketCleanupPolicy_concurrentUpdate(t *testing.T) { + var cleaner *EmptyFolderCleaner + calls := 0 + mock := &mockFilerOps{ + attrsFn: func(_ util.FullPath) (map[string][]byte, error) { + calls++ + if calls == 1 { + // a toggle lands while the stale read is in flight + cleaner.InvalidateBucketPolicy("/buckets", "test", true) + return map[string][]byte{s3_constants.ExtAllowEmptyFolders: []byte("true")}, nil + } + return map[string][]byte{s3_constants.ExtAllowEmptyFolders: []byte("false")}, nil + }, + } + + cleaner = &EmptyFolderCleaner{ + filer: mock, + bucketPath: "/buckets", + enabled: true, + bucketCleanupPolicies: make(map[string]*bucketCleanupPolicyState), + } + + _, autoRemove, _, attrValue, err := cleaner.getBucketCleanupPolicy(context.Background(), "/buckets/test/folder") + if err != nil { + t.Fatalf("getBucketCleanupPolicy: %v", err) + } + if !autoRemove || attrValue != "false" { + t.Fatalf("expected the post-update policy autoRemove=true attr=false, got autoRemove=%v attr=%q", autoRemove, attrValue) + } + if calls < 2 { + t.Fatalf("expected the stale load to be retried, got %d attribute reads", calls) + } + if cached := cleaner.bucketCleanupPolicies["/buckets/test"]; cached == nil || cached.attrValue != "false" { + t.Fatalf("expected the fresh policy to be cached, got %+v", cached) + } +} + +func TestEmptyFolderCleaner_getBucketCleanupPolicy_evictionKeepsInflightRead(t *testing.T) { + var cleaner *EmptyFolderCleaner + started := make(chan struct{}) + release := make(chan struct{}) + calls := 0 + mock := &mockFilerOps{ + attrsFn: func(_ util.FullPath) (map[string][]byte, error) { + calls++ + if calls == 1 { + close(started) + <-release + return map[string][]byte{s3_constants.ExtAllowEmptyFolders: []byte("true")}, nil + } + return map[string][]byte{s3_constants.ExtAllowEmptyFolders: []byte("false")}, nil + }, + } + + cleaner = &EmptyFolderCleaner{ + filer: mock, + bucketPath: "/buckets", + enabled: true, + cacheExpiry: time.Minute, + folderCounts: make(map[string]*folderState), + bucketCleanupPolicies: make(map[string]*bucketCleanupPolicyState), + policyGen: make(map[string]*bucketPolicyGen), + cleanupQueue: NewCleanupQueue(1000, time.Minute), + } + + go func() { + <-started + cleaner.InvalidateBucketPolicy("/buckets", "test", true) + cleaner.mu.Lock() + cleaner.policyGen["/buckets/test"].lastBump = time.Now().Add(-time.Hour) + cleaner.mu.Unlock() + cleaner.evictStaleCacheEntries() + close(release) + }() + + _, autoRemove, _, attrValue, err := cleaner.getBucketCleanupPolicy(context.Background(), "/buckets/test/folder") + if err != nil { + t.Fatalf("getBucketCleanupPolicy: %v", err) + } + if !autoRemove || attrValue != "false" { + t.Fatalf("expected the stalled read to be discarded, got autoRemove=%v attr=%q", autoRemove, attrValue) + } + if calls < 2 { + t.Fatalf("expected the stale load to be retried, got %d attribute reads", calls) + } +} + +func TestEmptyFolderCleaner_executeCleanup_policyFailureRequeues(t *testing.T) { + lockRing := lock_manager.NewLockRing(5 * time.Second) + lockRing.SetSnapshot([]pb.ServerAddress{"filer1:8888"}, 0) + + mock := &mockFilerOps{ + attrsFn: func(_ util.FullPath) (map[string][]byte, error) { + return nil, errors.New("attrs unavailable") + }, + } + + cleaner := &EmptyFolderCleaner{ + filer: mock, + lockRing: lockRing, + host: "filer1:8888", + bucketPath: "/buckets", + enabled: true, + folderCounts: make(map[string]*folderState), + bucketCleanupPolicies: make(map[string]*bucketCleanupPolicyState), + cleanupQueue: NewCleanupQueue(1000, time.Millisecond), + stopCh: make(chan struct{}), + } + + folder := "/buckets/test/folder" + for i := 0; i < DefaultMaxPolicyFailures; i++ { + cleaner.executeCleanup(folder, "triggered_item") + popped, _, ok := cleaner.cleanupQueue.Pop() + if !ok || popped != folder { + t.Fatalf("failure %d of %d should requeue the folder", i+1, DefaultMaxPolicyFailures) + } + } + + cleaner.executeCleanup(folder, "triggered_item") + + if got := cleaner.folderCounts[folder].policyFailures; got != DefaultMaxPolicyFailures+1 { + t.Fatalf("policyFailures = %d, want %d", got, DefaultMaxPolicyFailures+1) + } + if got := cleaner.cleanupQueue.Len(); got != 0 { + t.Fatalf("expected retries to stop after %d failures, got %d queued", DefaultMaxPolicyFailures, got) + } +} + func TestEmptyFolderCleaner_executeCleanup_directoryMarker(t *testing.T) { testCases := []struct { name string diff --git a/weed/filer/filer_on_meta_event.go b/weed/filer/filer_on_meta_event.go index d57810fac..264240548 100644 --- a/weed/filer/filer_on_meta_event.go +++ b/weed/filer/filer_on_meta_event.go @@ -55,11 +55,18 @@ func (f *Filer) onEmptyFolderCleanupEvents(event *filer_pb.SubscribeMetadataResp // Handle delete events - trigger folder cleanup check if filer_pb.IsDelete(event) && message.OldEntry != nil { f.EmptyFolderCleaner.OnDeleteEvent(directory, message.OldEntry.Name, message.OldEntry.IsDirectory, eventTime) + f.EmptyFolderCleaner.InvalidateBucketPolicy(directory, message.OldEntry.Name, message.OldEntry.IsDirectory) } // Handle create events - cancel pending cleanup for the folder if filer_pb.IsCreate(event) && message.NewEntry != nil { f.EmptyFolderCleaner.OnCreateEvent(directory, message.NewEntry.Name, message.NewEntry.IsDirectory) + f.EmptyFolderCleaner.InvalidateBucketPolicy(directory, message.NewEntry.Name, message.NewEntry.IsDirectory) + } + + // Handle update events - drop the cached bucket cleanup policy + if filer_pb.IsUpdate(event) && message.NewEntry != nil { + f.EmptyFolderCleaner.InvalidateBucketPolicy(directory, message.NewEntry.Name, message.NewEntry.IsDirectory) } // Handle rename/move events @@ -67,6 +74,7 @@ func (f *Filer) onEmptyFolderCleanupEvents(event *filer_pb.SubscribeMetadataResp // Treat the old location as a delete if message.OldEntry != nil { f.EmptyFolderCleaner.OnDeleteEvent(directory, message.OldEntry.Name, message.OldEntry.IsDirectory, eventTime) + f.EmptyFolderCleaner.InvalidateBucketPolicy(directory, message.OldEntry.Name, message.OldEntry.IsDirectory) } // Treat the new location as a create if message.NewEntry != nil { @@ -75,6 +83,7 @@ func (f *Filer) onEmptyFolderCleanupEvents(event *filer_pb.SubscribeMetadataResp newDir = directory } f.EmptyFolderCleaner.OnCreateEvent(newDir, message.NewEntry.Name, message.NewEntry.IsDirectory) + f.EmptyFolderCleaner.InvalidateBucketPolicy(newDir, message.NewEntry.Name, message.NewEntry.IsDirectory) } } } diff --git a/weed/pb/filer_pb/filer_pb_helper.go b/weed/pb/filer_pb/filer_pb_helper.go index 0985c440c..f818292c4 100644 --- a/weed/pb/filer_pb/filer_pb_helper.go +++ b/weed/pb/filer_pb/filer_pb_helper.go @@ -177,6 +177,22 @@ func CreateEntryWithResponse(ctx context.Context, client SeaweedFilerClient, req return resp, nil } +// SnapshotExtended clones extended attributes for use as an UpdateEntry +// ExpectedExtended precondition. Listed keys are asserted even when absent, +// so a concurrent add still fails the precondition. +func SnapshotExtended(extended map[string][]byte, keys ...string) map[string][]byte { + expected := make(map[string][]byte, len(extended)+len(keys)) + for k, v := range extended { + expected[k] = v + } + for _, k := range keys { + if _, ok := expected[k]; !ok { + expected[k] = nil + } + } + return expected +} + func UpdateEntry(ctx context.Context, client SeaweedFilerClient, request *UpdateEntryRequest) error { _, err := UpdateEntryWithResponse(ctx, client, request) return err diff --git a/weed/shell/command_s3_bucket_allow_empty_folders.go b/weed/shell/command_s3_bucket_allow_empty_folders.go new file mode 100644 index 000000000..b29f7f68a --- /dev/null +++ b/weed/shell/command_s3_bucket_allow_empty_folders.go @@ -0,0 +1,130 @@ +package shell + +import ( + "context" + "flag" + "fmt" + "io" + "strings" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3bucket" +) + +func init() { + Commands = append(Commands, &commandS3BucketAllowEmptyFolders{}) +} + +type commandS3BucketAllowEmptyFolders struct { +} + +func (c *commandS3BucketAllowEmptyFolders) Name() string { + return "s3.bucket.allowEmptyFolders" +} + +func (c *commandS3BucketAllowEmptyFolders) Help() string { + return `view or toggle whether a bucket keeps empty folders + + When enabled, the empty folder cleaner skips this bucket, so empty + directories persist (POSIX semantics). weed mount sets this when + mounting a bucket root. + + When disabled, folders emptied by object deletes are removed + asynchronously and stop appearing as CommonPrefix in S3 listings. + The explicit "false" marker is kept by weed mount. + + Example: + # Show the current setting + s3.bucket.allowEmptyFolders -name + + # Keep empty folders (skip the cleaner for this bucket) + s3.bucket.allowEmptyFolders -name -enable + + # Let the cleaner remove empty folders + s3.bucket.allowEmptyFolders -name -disable +` +} + +func (c *commandS3BucketAllowEmptyFolders) HasTag(CommandTag) bool { + return false +} + +func (c *commandS3BucketAllowEmptyFolders) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) { + bucketCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) + bucketName := bucketCommand.String("name", "", "bucket name") + enable := bucketCommand.Bool("enable", false, "keep empty folders in the bucket") + disable := bucketCommand.Bool("disable", false, "remove empty folders asynchronously") + if err = bucketCommand.Parse(args); err != nil { + return err + } + + if *bucketName == "" { + return fmt.Errorf("empty bucket name") + } + if err := s3bucket.VerifyS3BucketName(*bucketName); err != nil { + return fmt.Errorf("invalid bucket name %q: %w", *bucketName, err) + } + if *enable && *disable { + return fmt.Errorf("only one of -enable or -disable can be set") + } + + filerBucketsPath, err := readFilerBucketsPath(commandEnv) + if err != nil { + return fmt.Errorf("read buckets: %w", err) + } + + return commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + for attempt := 0; attempt < 5; attempt++ { + lookupResp, err := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{ + Directory: filerBucketsPath, + Name: *bucketName, + }) + if err != nil { + return fmt.Errorf("lookup bucket %s: %w", *bucketName, err) + } + entry := lookupResp.Entry + + if !*enable && !*disable { + state := "disabled" + if strings.EqualFold(strings.TrimSpace(string(entry.Extended[s3_constants.ExtAllowEmptyFolders])), "true") { + state = "enabled" + } + fmt.Fprintf(writer, "Bucket: %s\n", *bucketName) + fmt.Fprintf(writer, "Allow empty folders: %s\n", state) + return nil + } + + expected := filer_pb.SnapshotExtended(entry.Extended, s3_constants.ExtAllowEmptyFolders) + + if entry.Extended == nil { + entry.Extended = make(map[string][]byte) + } + state := "disabled" + if *enable { + entry.Extended[s3_constants.ExtAllowEmptyFolders] = []byte("true") + state = "enabled" + } else { + entry.Extended[s3_constants.ExtAllowEmptyFolders] = []byte("false") + } + + if _, err := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{ + Directory: filerBucketsPath, + Entry: entry, + ExpectedExtended: expected, + }); err != nil { + if status.Code(err) == codes.FailedPrecondition { + continue + } + return fmt.Errorf("failed to update bucket: %w", err) + } + + fmt.Fprintf(writer, "Bucket %s allow empty folders %s\n", *bucketName, state) + return nil + } + return fmt.Errorf("bucket %s changed concurrently; please retry", *bucketName) + }) +}