mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-19 05:36:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

S3: delay empty folder cleanup to prevent Spark write failures (#8970)

Browse Source 
* S3: delay empty folder cleanup to prevent Spark write failures (#8963)

Empty folders were being cleaned up within seconds, causing Apache Spark
(s3a) writes to fail when temporary directories like _temporary/0/task_xxx/
were briefly empty.

- Increase default cleanup delay from 5s to 2 minutes
- Only process queue items that have individually aged past the delay
  (previously the entire queue was drained once any item triggered)
- Make the delay configurable via filer.toml:
  [filer.options]
  s3.empty_folder_cleanup_delay = "2m"

* test: increase cleanup wait timeout to match 2m delay

The empty folder cleanup delay was increased to 2 minutes, so the
Spark integration test needs to wait longer for temporary directories
to disappear.

* fix: eagerly clean parent directories after empty folder deletion

After deleting an empty folder, immediately try to clean its parent
rather than relying on cascading metadata events that each re-enter
the 2-minute delay queue. This prevents multi-minute waits when
cleaning nested temporary directory trees (e.g. Spark's _temporary
hierarchy with 3+ levels would take 6m+ vs near-instant).

Fixes the CI failure where lingering _temporary parent directories
were not cleaned within the test's 3-minute timeout.
This commit is contained in:
Chris Lu
2026-04-07 13:20:59 -07:00
committed by GitHub
parent 761ec7da00
commit a4753b6a3b
7 changed files with 74 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/s3/spark/issue_8285_repro_test.go
+2 -1
View File
@@ -70,7 +70,8 @@ print("WRITE_COUNT=" + str(count))
"issue-8285/output/_temporary/0/",
"issue-8285/output/_temporary/0/_temporary/",
}
lingering := waitForObjectsToDisappear(t, env, "test", temporaryCandidates, 35*time.Second)
// Empty folder cleanup has a 2m default delay + 30s processor interval
lingering := waitForObjectsToDisappear(t, env, "test", temporaryCandidates, 3*time.Minute)
if len(lingering) > 0 {
t.Fatalf("issue #8285 regression detected: lingering temporary directories: %v", lingering)
}
weed/command/scaffold/filer.toml
+3
View File
@@ -13,6 +13,9 @@
# recursive_delete will delete all sub folders and files, similar to "rm -Rf"
recursive_delete = false
#max_file_name_length = 255
# for S3: how long to wait before deleting an empty folder.
# increase this if using tools like Spark that create temporary directories.
#s3.empty_folder_cleanup_delay = "2m"
####################################################
# The following are filer store options
weed/filer/empty_folder_cleanup/cleanup_queue.go
+21
View File
@@ -153,6 +153,27 @@ func (q *CleanupQueue) Pop() (string, string, bool) {
return item.folder, item.triggeredBy, true
}
// PopOlderThan removes and returns the oldest folder only if it has been in the queue
// for longer than the specified duration. Returns empty string and false if no item qualifies.
func (q *CleanupQueue) PopOlderThan(minAge time.Duration) (string, string, bool) {
q.mu.Lock()
defer q.mu.Unlock()
front := q.items.Front()
if front == nil {
return "", "", false
}
item := front.Value.(*queueItem)
if time.Since(item.queueTime) <= minAge {
return "", "", false
}
q.items.Remove(front)
delete(q.itemsMap, item.folder)
return item.folder, item.triggeredBy, true
}
// Peek returns the oldest folder without removing it.
// Returns the folder and queue time if available, or empty values if queue is empty.
func (q *CleanupQueue) Peek() (folder string, triggeredBy string, queueTime time.Time, ok bool) {
weed/filer/empty_folder_cleanup/empty_folder_cleaner.go
+28 -19
View File
@@ -18,8 +18,8 @@ const (
DefaultMaxCountCheck = 1000
DefaultCacheExpiry = 5 * time.Minute
DefaultQueueMaxSize = 1000
DefaultQueueMaxAge = 5 * time.Second
DefaultProcessorSleep = 10 * time.Second // How often to check queue
DefaultQueueMaxAge = 2 * time.Minute
DefaultProcessorSleep = 30 * time.Second // How often to check queue
)
// FilerOperations defines the filer operations needed by EmptyFolderCleaner
@@ -70,15 +70,20 @@ type EmptyFolderCleaner struct {
stopCh chan struct{}
}
// NewEmptyFolderCleaner creates a new EmptyFolderCleaner
func NewEmptyFolderCleaner(filer FilerOperations, lockRing *lock_manager.LockRing, host pb.ServerAddress, bucketPath string) *EmptyFolderCleaner {
// NewEmptyFolderCleaner creates a new EmptyFolderCleaner.
// cleanupDelay controls how long an empty folder must remain in the queue before deletion.
// If zero, DefaultQueueMaxAge is used.
func NewEmptyFolderCleaner(filer FilerOperations, lockRing *lock_manager.LockRing, host pb.ServerAddress, bucketPath string, cleanupDelay time.Duration) *EmptyFolderCleaner {
if cleanupDelay <= 0 {
cleanupDelay = DefaultQueueMaxAge
}
efc := &EmptyFolderCleaner{
filer: filer,
lockRing: lockRing,
host: host,
folderCounts: make(map[string]*folderState),
bucketCleanupPolicies: make(map[string]*bucketCleanupPolicyState),
cleanupQueue: NewCleanupQueue(DefaultQueueMaxSize, DefaultQueueMaxAge),
cleanupQueue: NewCleanupQueue(DefaultQueueMaxSize, cleanupDelay),
maxCountCheck: DefaultMaxCountCheck,
cacheExpiry: DefaultCacheExpiry,
processorSleep: DefaultProcessorSleep,
@@ -207,27 +212,22 @@ func (efc *EmptyFolderCleaner) cleanupProcessor() {
// processCleanupQueue processes items from the cleanup queue
func (efc *EmptyFolderCleaner) processCleanupQueue() {
// Check if we should process
if !efc.cleanupQueue.ShouldProcess() {
if efc.cleanupQueue.Len() > 0 {
glog.Infof("EmptyFolderCleaner: pending queue not processed yet (len=%d, oldest_age=%v, max_size=%d, max_age=%v)",
efc.cleanupQueue.Len(), efc.cleanupQueue.OldestAge(), efc.cleanupQueue.maxSize, efc.cleanupQueue.maxAge)
}
if efc.cleanupQueue.Len() == 0 {
return
}
glog.V(3).Infof("EmptyFolderCleaner: processing cleanup queue (len=%d, age=%v)",
glog.V(3).Infof("EmptyFolderCleaner: processing cleanup queue (len=%d, oldest_age=%v)",
efc.cleanupQueue.Len(), efc.cleanupQueue.OldestAge())
// Process all items that are ready
for efc.cleanupQueue.Len() > 0 {
// Only process items that have been queued longer than maxAge
for {
// Check if still enabled
if !efc.IsEnabled() {
return
}
// Pop the oldest item
folder, triggeredBy, ok := efc.cleanupQueue.Pop()
// Only pop items old enough — newer items stay in the queue
folder, triggeredBy, ok := efc.cleanupQueue.PopOlderThan(efc.cleanupQueue.maxAge)
if !ok {
break
}
@@ -322,9 +322,18 @@ func (efc *EmptyFolderCleaner) executeCleanup(folder string, triggeredBy string)
delete(efc.folderCounts, folder)
efc.mu.Unlock()
// Note: No need to recursively check parent folder here.
// The deletion of this folder will generate a metadata event,
// which will trigger OnDeleteEvent for the parent folder.
// After deleting this folder, immediately try to clean the parent.
// Relying solely on cascading metadata events would re-enter the full
// delay queue for each ancestor level, causing multi-minute cascading
// waits (e.g. 3 levels × 2m = 6m+). Instead, walk up eagerly.
parentDir, _ := util.FullPath(folder).DirAndName()
if parentDir != "" && parentDir != folder &&
efc.bucketPath != "" && isUnderBucketPath(parentDir, efc.bucketPath) {
// Remove any pending queue entry for the parent so we don't
// double-process it later from a stale event.
efc.cleanupQueue.Remove(parentDir)
efc.executeCleanup(parentDir, triggeredBy)
}
}
// countItems counts items in a folder (up to maxCountCheck)
weed/filer/empty_folder_cleanup/empty_folder_cleaner_test.go
+13 -10
View File
@@ -655,7 +655,7 @@ func TestEmptyFolderCleaner_queueFIFOOrder(t *testing.T) {
cleaner.Stop()
}
func TestEmptyFolderCleaner_processCleanupQueue_drainsAllOnceTriggered(t *testing.T) {
func TestEmptyFolderCleaner_processCleanupQueue_onlyProcessesAgedItems(t *testing.T) {
lockRing := lock_manager.NewLockRing(5 * time.Second)
lockRing.SetSnapshot([]pb.ServerAddress{"filer1:8888"}, 0)
@@ -670,6 +670,7 @@ func TestEmptyFolderCleaner_processCleanupQueue_drainsAllOnceTriggered(t *testin
},
}
maxAge := 100 * time.Millisecond
cleaner := &EmptyFolderCleaner{
filer: mock,
lockRing: lockRing,
@@ -677,25 +678,27 @@ func TestEmptyFolderCleaner_processCleanupQueue_drainsAllOnceTriggered(t *testin
bucketPath: "/buckets",
enabled: true,
folderCounts: make(map[string]*folderState),
cleanupQueue: NewCleanupQueue(2, time.Hour),
cleanupQueue: NewCleanupQueue(1000, maxAge),
maxCountCheck: 1000,
cacheExpiry: time.Minute,
processorSleep: time.Second,
stopCh: make(chan struct{}),
}
now := time.Now()
cleaner.cleanupQueue.Add("/buckets/test/folder1", "i1", now)
cleaner.cleanupQueue.Add("/buckets/test/folder2", "i2", now.Add(time.Millisecond))
cleaner.cleanupQueue.Add("/buckets/test/folder3", "i3", now.Add(2*time.Millisecond))
// Add old items (well past maxAge) and a fresh item
old := time.Now().Add(-time.Second)
cleaner.cleanupQueue.Add("/buckets/test/folder1", "i1", old)
cleaner.cleanupQueue.Add("/buckets/test/folder2", "i2", old.Add(time.Millisecond))
cleaner.cleanupQueue.Add("/buckets/test/folder3", "i3", time.Now()) // fresh, should NOT be processed
cleaner.processCleanupQueue()
if got := cleaner.cleanupQueue.Len(); got != 0 {
t.Fatalf("expected queue to be drained, got len=%d", got)
// Only the two old items should have been processed
if len(deleted) != 2 {
t.Fatalf("expected 2 deleted folders (aged items only), got %d: %v", len(deleted), deleted)
}
if len(deleted) != 3 {
t.Fatalf("expected 3 deleted folders, got %d", len(deleted))
if got := cleaner.cleanupQueue.Len(); got != 1 {
t.Fatalf("expected 1 item remaining in queue, got %d", got)
}
}
weed/filer/filer.go
+3 -2
View File
@@ -61,7 +61,8 @@ type Filer struct {
MaxFilenameLength uint32
deletionQuit chan struct{}
DeletionRetryQueue *DeletionRetryQueue
EmptyFolderCleaner *empty_folder_cleanup.EmptyFolderCleaner
EmptyFolderCleaner *empty_folder_cleanup.EmptyFolderCleaner
EmptyFolderCleanupDelay time.Duration
}
func NewFiler(masters pb.ServerDiscovery, grpcDialOption grpc.DialOption, filerHost pb.ServerAddress, filerGroup string, collection string, replication string, dataCenter string, maxFilenameLength uint32, notifyFn func()) *Filer {
@@ -123,7 +124,7 @@ func (f *Filer) AggregateFromPeers(self pb.ServerAddress, existingNodes []*maste
glog.V(0).Infof("%s aggregate from peers %+v", self, snapshot)
// Initialize the empty folder cleaner using the same LockRing as Dlm for consistent hashing
f.EmptyFolderCleaner = empty_folder_cleanup.NewEmptyFolderCleaner(f, f.Dlm.LockRing, self, f.DirBucketsPath)
f.EmptyFolderCleaner = empty_folder_cleanup.NewEmptyFolderCleaner(f, f.Dlm.LockRing, self, f.DirBucketsPath, f.EmptyFolderCleanupDelay)
f.MetaAggregator = NewMetaAggregator(f, self, f.GrpcDialOption)
f.MasterClient.SetOnPeerUpdateFn(func(update *master_pb.ClusterNodeUpdate, startFrom time.Time) {
weed/server/filer_server.go
+4
View File
@@ -248,6 +248,10 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
glog.Fatalf("%s bootstrap from %+v: %v", option.Host, existingNodes, err)
}
}
v.SetDefault("filer.options.s3.empty_folder_cleanup_delay", "2m")
if d, err := time.ParseDuration(v.GetString("filer.options.s3.empty_folder_cleanup_delay")); err == nil {
fs.filer.EmptyFolderCleanupDelay = d
}
fs.filer.AggregateFromPeers(option.Host, existingNodes, startFromTime)
fs.filer.LoadFilerConf()