From 0d7173a0291f36c50b4912b8c793a446c936d437 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 1 Aug 2026 20:11:41 -0700 Subject: [PATCH] remote storage: actually delete objects when a directory is removed (#10531) * remote storage: actually delete objects when a directory is removed On object-store backends RemoveDirectory returned nil without doing anything, so a directory delete synced to the remote as a successful no-op and the objects under that prefix stayed there forever. Nothing surfaced the divergence: the sync logged rmdir, advanced its offset, and the local namespace looked clean. Deleting a bucket-level directory on a filer store that can drop a whole bucket emits no per-child delete events at all, so the single rmdir event was the only chance to clean up the remote. Each backend now lists the prefix and deletes what it finds: S3 in DeleteObjects batches of one listing page, GCS and Azure per object. The prefix always ends with a slash so a sibling like dir2 survives deleting dir, and errors propagate so a failed delete is retried instead of silently skipped. A directory that maps to the bucket root is left alone: wiping every object in the bucket from one namespace event is too destructive, and bucket removal already has its own path. * gcs remote: wrap the per-object delete error The listing error in the same function already wraps, so the delete error should stay inspectable with errors.Is as well. * s3 remote: name the empty-listing test for what it checks The prefix in that test is a normal directory; what is empty is the listing. The bucket-root guard has its own test. * s3 remote: report the scope of a failed delete batch A DeleteObjects response can carry per-key errors for up to a thousand keys. Surfacing only the first hid how much of the batch failed, and surfacing all of them would build an unbounded error string, so report the count with the first failure as the sample. --- .../azure/azure_storage_client.go | 34 +++++ weed/remote_storage/gcs/gcs_storage_client.go | 29 ++++- weed/remote_storage/s3/s3_storage_client.go | 51 ++++++++ .../s3/s3_storage_client_test.go | 122 ++++++++++++++++++ 4 files changed, 235 insertions(+), 1 deletion(-) diff --git a/weed/remote_storage/azure/azure_storage_client.go b/weed/remote_storage/azure/azure_storage_client.go index 705a5720a..5e9495ab2 100644 --- a/weed/remote_storage/azure/azure_storage_client.go +++ b/weed/remote_storage/azure/azure_storage_client.go @@ -19,6 +19,7 @@ import ( "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/bloberror" "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blockblob" "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container" + "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" "github.com/seaweedfs/seaweedfs/weed/remote_storage" @@ -382,6 +383,39 @@ func (az *azureRemoteStorageClient) WriteDirectory(loc *remote_pb.RemoteStorageL } func (az *azureRemoteStorageClient) RemoveDirectory(loc *remote_pb.RemoteStorageLocation) (err error) { + // the trailing slash keeps sibling prefixes that share the name intact + prefix := loc.Path[1:] + if prefix != "" && !strings.HasSuffix(prefix, "/") { + prefix += "/" + } + if prefix == "" { + // the mount root maps to the whole container; wiping every blob from a + // single namespace event is too destructive, so keep them + glog.Warningf("azure %s: skip removing directory mapped to the container root", loc.Bucket) + return nil + } + + containerClient := az.client.ServiceClient().NewContainerClient(loc.Bucket) + pager := containerClient.NewListBlobsFlatPager(&container.ListBlobsFlatOptions{ + Prefix: &prefix, + }) + for pager.More() { + resp, pageErr := pager.NextPage(context.Background()) + if pageErr != nil { + return fmt.Errorf("azure list %s/%s: %w", loc.Bucket, prefix, pageErr) + } + for _, blobItem := range resp.Segment.BlobItems { + if blobItem.Name == nil { + continue + } + _, delErr := containerClient.NewBlobClient(*blobItem.Name).Delete(context.Background(), &blob.DeleteOptions{ + DeleteSnapshots: to.Ptr(blob.DeleteSnapshotsOptionTypeInclude), + }) + if delErr != nil && !bloberror.HasCode(delErr, bloberror.BlobNotFound) { + return fmt.Errorf("azure delete %s/%s: %w", loc.Bucket, *blobItem.Name, delErr) + } + } + } return nil } diff --git a/weed/remote_storage/gcs/gcs_storage_client.go b/weed/remote_storage/gcs/gcs_storage_client.go index f776ae527..4b535d514 100644 --- a/weed/remote_storage/gcs/gcs_storage_client.go +++ b/weed/remote_storage/gcs/gcs_storage_client.go @@ -220,7 +220,34 @@ func (gcs *gcsRemoteStorageClient) WriteDirectory(loc *remote_pb.RemoteStorageLo } func (gcs *gcsRemoteStorageClient) RemoveDirectory(loc *remote_pb.RemoteStorageLocation) (err error) { - return nil + // the trailing slash keeps sibling prefixes that share the name intact + prefix := loc.Path[1:] + if prefix != "" && !strings.HasSuffix(prefix, "/") { + prefix += "/" + } + if prefix == "" { + // the mount root maps to the whole bucket; wiping every object from a + // single namespace event is too destructive, so keep them + glog.Warningf("gcs %s: skip removing directory mapped to the bucket root", loc.Bucket) + return nil + } + + bucket := gcs.client.Bucket(loc.Bucket) + objectIterator := bucket.Objects(context.Background(), &storage.Query{ + Prefix: prefix, + }) + for { + objectAttr, iterErr := objectIterator.Next() + if iterErr == iterator.Done { + return nil + } + if iterErr != nil { + return fmt.Errorf("gcs list %s/%s: %w", loc.Bucket, prefix, iterErr) + } + if delErr := bucket.Object(objectAttr.Name).Delete(context.Background()); delErr != nil && !errors.Is(delErr, storage.ErrObjectNotExist) { + return fmt.Errorf("gcs delete %s/%s: %w", loc.Bucket, objectAttr.Name, delErr) + } + } } func (gcs *gcsRemoteStorageClient) WriteFile(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry, reader io.Reader) (remoteEntry *filer_pb.RemoteEntry, err error) { diff --git a/weed/remote_storage/s3/s3_storage_client.go b/weed/remote_storage/s3/s3_storage_client.go index 7505676cb..1132b36d3 100644 --- a/weed/remote_storage/s3/s3_storage_client.go +++ b/weed/remote_storage/s3/s3_storage_client.go @@ -285,6 +285,57 @@ func (s *s3RemoteStorageClient) WriteDirectory(loc *remote_pb.RemoteStorageLocat } func (s *s3RemoteStorageClient) RemoveDirectory(loc *remote_pb.RemoteStorageLocation) (err error) { + // the trailing slash keeps sibling prefixes that share the name intact + prefix := loc.Path[1:] + if prefix != "" && !strings.HasSuffix(prefix, "/") { + prefix += "/" + } + if prefix == "" { + // the mount root maps to the whole bucket; wiping every object from a + // single namespace event is too destructive, so keep them + glog.Warningf("s3 %s: skip removing directory mapped to the bucket root", loc.Bucket) + return nil + } + + listInput := &s3.ListObjectsV2Input{ + Bucket: aws.String(loc.Bucket), + Prefix: aws.String(prefix), + } + var deleteErr error + listErr := s.conn.ListObjectsV2Pages(listInput, func(page *s3.ListObjectsV2Output, lastPage bool) bool { + var objects []*s3.ObjectIdentifier + for _, content := range page.Contents { + objects = append(objects, &s3.ObjectIdentifier{Key: content.Key}) + } + if len(objects) == 0 { + return true + } + // a listing page holds at most 1000 keys, the DeleteObjects limit + resp, batchErr := s.conn.DeleteObjects(&s3.DeleteObjectsInput{ + Bucket: aws.String(loc.Bucket), + Delete: &s3.Delete{ + Objects: objects, + Quiet: aws.Bool(true), + }, + }) + if batchErr != nil { + deleteErr = batchErr + return false + } + if len(resp.Errors) > 0 { + // a batch can fail 1000 keys; report the scope, not every key + failed := resp.Errors[0] + deleteErr = fmt.Errorf("%d keys failed, first is %s: %s %s", len(resp.Errors), aws.StringValue(failed.Key), aws.StringValue(failed.Code), aws.StringValue(failed.Message)) + return false + } + return true + }) + if listErr != nil { + return fmt.Errorf("list %s/%s: %w", loc.Bucket, prefix, listErr) + } + if deleteErr != nil { + return fmt.Errorf("remove directory %s/%s: %w", loc.Bucket, prefix, deleteErr) + } return nil } diff --git a/weed/remote_storage/s3/s3_storage_client_test.go b/weed/remote_storage/s3/s3_storage_client_test.go index 79bb794b6..1ac666063 100644 --- a/weed/remote_storage/s3/s3_storage_client_test.go +++ b/weed/remote_storage/s3/s3_storage_client_test.go @@ -2,13 +2,16 @@ package s3 import ( "bytes" + "fmt" "io" "net/http" "strings" "testing" + "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/credentials" awss3 "github.com/aws/aws-sdk-go/service/s3" + "github.com/aws/aws-sdk-go/service/s3/s3iface" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" "github.com/seaweedfs/seaweedfs/weed/remote_storage" @@ -71,6 +74,125 @@ func TestS3ErrRemoteObjectNotFoundIsAccessible(t *testing.T) { require.Equal(t, "remote object not found", remote_storage.ErrRemoteObjectNotFound.Error()) } +// removeDirectoryMock serves canned listing pages and records the delete batches. +type removeDirectoryMock struct { + s3iface.S3API + pages []*awss3.ListObjectsV2Output + listInputs []*awss3.ListObjectsV2Input + deleteInputs []*awss3.DeleteObjectsInput + deleteResp *awss3.DeleteObjectsOutput + deleteErr error +} + +func (m *removeDirectoryMock) ListObjectsV2Pages(input *awss3.ListObjectsV2Input, fn func(*awss3.ListObjectsV2Output, bool) bool) error { + m.listInputs = append(m.listInputs, input) + for i, page := range m.pages { + if !fn(page, i == len(m.pages)-1) { + return nil + } + } + return nil +} + +func (m *removeDirectoryMock) DeleteObjects(input *awss3.DeleteObjectsInput) (*awss3.DeleteObjectsOutput, error) { + m.deleteInputs = append(m.deleteInputs, input) + if m.deleteErr != nil { + return nil, m.deleteErr + } + if m.deleteResp != nil { + return m.deleteResp, nil + } + return &awss3.DeleteObjectsOutput{}, nil +} + +func listPage(keys ...string) *awss3.ListObjectsV2Output { + page := &awss3.ListObjectsV2Output{} + for _, key := range keys { + page.Contents = append(page.Contents, &awss3.Object{Key: aws.String(key)}) + } + return page +} + +func deletedKeys(input *awss3.DeleteObjectsInput) (keys []string) { + for _, object := range input.Delete.Objects { + keys = append(keys, aws.StringValue(object.Key)) + } + return +} + +func TestS3RemoveDirectoryDeletesEveryListedObject(t *testing.T) { + mock := &removeDirectoryMock{ + pages: []*awss3.ListObjectsV2Output{ + listPage("testdir/a.bin", "testdir/sub/b.bin"), + listPage("testdir/z.bin"), + }, + } + client := &s3RemoteStorageClient{conf: &remote_pb.RemoteConf{Name: "test"}, conn: mock} + loc := &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/testdir"} + + require.NoError(t, client.RemoveDirectory(loc)) + + require.Len(t, mock.listInputs, 1) + // without the trailing slash the listing would also match /testdir2 + require.Equal(t, "testdir/", aws.StringValue(mock.listInputs[0].Prefix)) + require.Len(t, mock.deleteInputs, 2) + require.Equal(t, []string{"testdir/a.bin", "testdir/sub/b.bin"}, deletedKeys(mock.deleteInputs[0])) + require.Equal(t, []string{"testdir/z.bin"}, deletedKeys(mock.deleteInputs[1])) +} + +func TestS3RemoveDirectoryEmptyListingSendsNoDeletes(t *testing.T) { + mock := &removeDirectoryMock{pages: []*awss3.ListObjectsV2Output{listPage()}} + client := &s3RemoteStorageClient{conf: &remote_pb.RemoteConf{Name: "test"}, conn: mock} + loc := &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/testdir"} + + require.NoError(t, client.RemoveDirectory(loc)) + require.Empty(t, mock.deleteInputs) +} + +func TestS3RemoveDirectoryRefusesBucketRoot(t *testing.T) { + mock := &removeDirectoryMock{pages: []*awss3.ListObjectsV2Output{listPage("a.bin")}} + client := &s3RemoteStorageClient{conf: &remote_pb.RemoteConf{Name: "test"}, conn: mock} + loc := &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/"} + + require.NoError(t, client.RemoveDirectory(loc)) + require.Empty(t, mock.listInputs) + require.Empty(t, mock.deleteInputs) +} + +func TestS3RemoveDirectoryReturnsBatchError(t *testing.T) { + mock := &removeDirectoryMock{ + pages: []*awss3.ListObjectsV2Output{listPage("testdir/a.bin")}, + deleteErr: fmt.Errorf("access denied"), + } + client := &s3RemoteStorageClient{conf: &remote_pb.RemoteConf{Name: "test"}, conn: mock} + loc := &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/testdir"} + + require.ErrorContains(t, client.RemoveDirectory(loc), "access denied") +} + +func TestS3RemoveDirectoryReturnsPerKeyError(t *testing.T) { + mock := &removeDirectoryMock{ + pages: []*awss3.ListObjectsV2Output{listPage("testdir/a.bin", "testdir/b.bin")}, + deleteResp: &awss3.DeleteObjectsOutput{ + Errors: []*awss3.Error{{ + Key: aws.String("testdir/a.bin"), + Code: aws.String("InternalError"), + Message: aws.String("try again"), + }, { + Key: aws.String("testdir/b.bin"), + Code: aws.String("InternalError"), + Message: aws.String("try again"), + }}, + }, + } + client := &s3RemoteStorageClient{conf: &remote_pb.RemoteConf{Name: "test"}, conn: mock} + loc := &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/testdir"} + + err := client.RemoveDirectory(loc) + require.ErrorContains(t, err, "2 keys failed") + require.ErrorContains(t, err, "testdir/a.bin") +} + // captureRoundTripper records the PUT request that the s3manager uploader // sends, and short-circuits all calls with a 200 so the SDK is satisfied. type captureRoundTripper struct {