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 {