diff --git a/weed/command/filer_remote_gateway_buckets.go b/weed/command/filer_remote_gateway_buckets.go index e07f70e877..db4b276a98 100644 --- a/weed/command/filer_remote_gateway_buckets.go +++ b/weed/command/filer_remote_gateway_buckets.go @@ -158,6 +158,11 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour client, remoteStorageMountLocation, err := option.findRemoteStorageClient(entry.Name) if err != nil { + bucket := util.FullPath(option.bucketsDir).Child(entry.Name) + if _, isMounted := option.mappings.Mappings[string(bucket)]; !isMounted { + glog.V(0).Infof("remote bucket %s is not mounted, skipping delete", entry.Name) + return nil + } return fmt.Errorf("findRemoteStorageClient %s: %v", entry.Name, err) } @@ -180,6 +185,9 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour if readErr != nil { return fmt.Errorf("unmarshal mappings: %w", readErr) } + if newMappings.PrimaryBucketStorageName == "" { + newMappings.PrimaryBucketStorageName = option.mappings.PrimaryBucketStorageName + } option.mappings = newMappings } if strings.HasSuffix(message.NewEntry.Name, filer.REMOTE_STORAGE_CONF_SUFFIX) { @@ -192,7 +200,10 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour } else if metadataEventRemovesFromDirectory(resp, filer.DirectoryEtcRemote) { // deletion if message.OldEntry.Name == filer.REMOTE_STORAGE_MOUNT_FILE { - option.mappings = &remote_pb.RemoteStorageMapping{} + option.mappings = &remote_pb.RemoteStorageMapping{ + PrimaryBucketStorageName: option.mappings.PrimaryBucketStorageName, + Mappings: map[string]*remote_pb.RemoteStorageLocation{}, + } } if strings.HasSuffix(message.OldEntry.Name, filer.REMOTE_STORAGE_CONF_SUFFIX) { conf := &remote_pb.RemoteConf{} @@ -331,7 +342,7 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour return client.RemoveDirectory(dest) } glog.V(0).Infof("delete %s", remote_storage.FormatLocation(dest)) - return client.DeleteFile(dest) + return deleteRemoteFile(client, dest) } if message.OldEntry != nil && message.NewEntry != nil { if resp.Directory == option.bucketsDir { diff --git a/weed/command/filer_remote_gateway_buckets_test.go b/weed/command/filer_remote_gateway_buckets_test.go index b6d36decd9..253b765189 100644 --- a/weed/command/filer_remote_gateway_buckets_test.go +++ b/weed/command/filer_remote_gateway_buckets_test.go @@ -340,3 +340,145 @@ func TestGatewayMetadataOnlyDeleteKeepsTheRemoteObject(t *testing.T) { t.Fatalf("deletes = %+v, want the ordinary delete", remote.deletes) } } + +func TestGatewayUnmountedBucketDeleteIsSkipped(t *testing.T) { + remote := &recordingRemote{} + remote_storage.RemoteStorageClientMakers["gatewaytest"] = recordingRemoteMaker{client: remote} + t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest") }) + option := &RemoteGatewayOptions{ + bucketsDir: "/buckets", + mappings: &remote_pb.RemoteStorageMapping{Mappings: map[string]*remote_pb.RemoteStorageLocation{}}, + remoteConfs: map[string]*remote_pb.RemoteConf{"gatewaytest": {Name: "gatewaytest", Type: "gatewaytest"}}, + } + process, err := option.makeBucketedEventProcessor(nil) + if err != nil { + t.Fatal(err) + } + + deleteBucketEvent := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{ + Name: "unmounted-bucket", + IsDirectory: true, + }, + }, + } + + // Deleting a bucket that is not mounted (for instance because it was already unmounted + // by an earlier copy of the delete event) should return nil and not error. + if err := process(deleteBucketEvent); err != nil { + t.Fatalf("expected nil for unmounted bucket delete, got: %v", err) + } +} + +// TestGatewayCreatesABucketAfterTheMappingFileIsDeleted: deleting +// mount.mapping empties the mapping table, and a later bucket create still +// records its mount in it. +func TestGatewayCreatesABucketAfterTheMappingFileIsDeleted(t *testing.T) { + option, remote := gatewayTestOption("gatewaytest-mapping-reset") + t.Cleanup(func() { + delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-mapping-reset") + remote_storage.ResetRemoteStorageClients() + }) + createAt, noFilter, noSuffix := "gatewaytest-mapping-reset", "", false + option.createBucketAt, option.include, option.exclude, option.createBucketRandomSuffix = &createAt, &noFilter, &noFilter, &noSuffix + process, err := option.makeBucketedEventProcessor(nil) + if err != nil { + t.Fatal(err) + } + + if err := process(&filer_pb.SubscribeMetadataResponse{ + Directory: filer.DirectoryEtcRemote, + EventNotification: &filer_pb.EventNotification{OldEntry: &filer_pb.Entry{Name: filer.REMOTE_STORAGE_MOUNT_FILE}}, + }); err != nil { + t.Fatal(err) + } + // saving the new mount needs a filer, so the create itself errors + _ = process(&filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets", + EventNotification: &filer_pb.EventNotification{ + NewParentPath: "/buckets", + NewEntry: &filer_pb.Entry{Name: "fresh", IsDirectory: true}, + }, + }) + + if !slices.Equal(remote.bucketCreates, []string{"fresh"}) { + t.Errorf("bucket creates = %v, want [fresh]", remote.bucketCreates) + } + if loc := option.mappings.Mappings["/buckets/fresh"]; loc.GetBucket() != "fresh" { + t.Errorf("mapping for /buckets/fresh = %v, want bucket fresh", loc) + } +} + +// TestGatewayDeleteOfAnAbsentObjectSucceeds: a repeated delete finds the +// object gone, which GCS reports as ErrRemoteObjectNotFound; the event has +// nothing left to do and must not hold the sync offset. +func TestGatewayDeleteOfAnAbsentObjectSucceeds(t *testing.T) { + option, remote := gatewayTestOption("gatewaytest-absent-delete") + t.Cleanup(func() { + delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-absent-delete") + remote_storage.ResetRemoteStorageClients() + }) + remote.deleteErr = remote_storage.ErrRemoteObjectNotFound + process, err := option.makeBucketedEventProcessor(nil) + if err != nil { + t.Fatal(err) + } + + if err := process(&filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/gold/dir", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "obj.bin", Attributes: &filer_pb.FuseAttributes{}}, + }, + }); err != nil { + t.Fatalf("delete of an absent object: %v", err) + } + if len(remote.deletes) != 1 || remote.deletes[0].Path != "/dir/obj.bin" { + t.Errorf("deletes = %+v, want /dir/obj.bin", remote.deletes) + } +} + +// TestGatewayKeepsTheDefaultStorageAcrossMappingChanges: the default storage +// for new buckets is derived at startup and never saved in mount.mapping, so +// replacing or deleting that file must not drop it. +func TestGatewayKeepsTheDefaultStorageAcrossMappingChanges(t *testing.T) { + for _, tt := range []struct { + name string + event *filer_pb.EventNotification + }{ + {"mapping updated", &filer_pb.EventNotification{NewParentPath: filer.DirectoryEtcRemote, NewEntry: &filer_pb.Entry{Name: filer.REMOTE_STORAGE_MOUNT_FILE}}}, + {"mapping deleted", &filer_pb.EventNotification{OldEntry: &filer_pb.Entry{Name: filer.REMOTE_STORAGE_MOUNT_FILE}}}, + } { + t.Run(tt.name, func(t *testing.T) { + option, remote := gatewayTestOption("gatewaytest-default-storage") + t.Cleanup(func() { + delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-default-storage") + remote_storage.ResetRemoteStorageClients() + }) + option.mappings.PrimaryBucketStorageName = "gatewaytest-default-storage" + createAt, noFilter, noSuffix := "", "", false + option.createBucketAt, option.include, option.exclude, option.createBucketRandomSuffix = &createAt, &noFilter, &noFilter, &noSuffix + process, err := option.makeBucketedEventProcessor(nil) + if err != nil { + t.Fatal(err) + } + + if err := process(&filer_pb.SubscribeMetadataResponse{Directory: filer.DirectoryEtcRemote, EventNotification: tt.event}); err != nil { + t.Fatal(err) + } + // saving the new mount needs a filer, so the create itself errors + _ = process(&filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets", + EventNotification: &filer_pb.EventNotification{ + NewParentPath: "/buckets", + NewEntry: &filer_pb.Entry{Name: "fresh", IsDirectory: true}, + }, + }) + + if !slices.Equal(remote.bucketCreates, []string{"fresh"}) { + t.Errorf("bucket creates = %v, want [fresh] in the default storage", remote.bucketCreates) + } + }) + } +} diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index a57c8c48b8..69371c25b4 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -884,6 +884,7 @@ type recordingRemote struct { // metadataUpdateResult is the RemoteEntry UpdateFileMetadata reports; nil // means the update left the object untouched metadataUpdateResult *filer_pb.RemoteEntry + bucketCreates []string } func (r *recordingRemote) WriteFile(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry, reader io.Reader) (*filer_pb.RemoteEntry, error) { @@ -932,6 +933,11 @@ func (r *recordingRemote) DeleteFile(loc *remote_pb.RemoteStorageLocation) error return r.deleteErr } +func (r *recordingRemote) CreateBucket(name string) error { + r.bucketCreates = append(r.bucketCreates, name) + return nil +} + func (r *recordingRemote) UpdateFileMetadata(loc *remote_pb.RemoteStorageLocation, oldEntry, newEntry *filer_pb.Entry) (*filer_pb.RemoteEntry, error) { r.metadataUpdates = append(r.metadataUpdates, loc) return r.metadataUpdateResult, nil diff --git a/weed/remote_storage/azure/azure_storage_client.go b/weed/remote_storage/azure/azure_storage_client.go index 15ff2519d1..bbb893c31a 100644 --- a/weed/remote_storage/azure/azure_storage_client.go +++ b/weed/remote_storage/azure/azure_storage_client.go @@ -645,6 +645,9 @@ func (az *azureRemoteStorageClient) DeleteBucket(name string) (err error) { containerClient := az.client.ServiceClient().NewContainerClient(name) _, err = containerClient.Delete(context.Background(), nil) if err != nil { + if bloberror.HasCode(err, bloberror.ContainerNotFound, bloberror.ContainerBeingDeleted) { + return nil + } return fmt.Errorf("delete bucket %s: %w", name, err) } return diff --git a/weed/remote_storage/azure/azure_storage_client_test.go b/weed/remote_storage/azure/azure_storage_client_test.go index 5c58a7bd73..2228eb0b64 100644 --- a/weed/remote_storage/azure/azure_storage_client_test.go +++ b/weed/remote_storage/azure/azure_storage_client_test.go @@ -483,3 +483,26 @@ func TestAzureReadNotFoundClassification(t *testing.T) { }) } } + +func TestAzureDeleteBucketOfAMissingContainerSucceeds(t *testing.T) { + tests := []struct { + name string + statusCode int + code string + wantErr bool + }{ + {"missing container", http.StatusNotFound, "ContainerNotFound", false}, + {"container being deleted", http.StatusConflict, "ContainerBeingDeleted", false}, + {"auth failure", http.StatusForbidden, "AuthenticationFailed", true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + conf := &remote_pb.RemoteConf{Name: "test", AzureAccountName: "testaccount", AzureAccountKey: "dGVzdGtleQ=="} + rs, err := MakeWithHTTPClient(conf, &http.Client{Transport: &azureErrorRoundTripper{statusCode: tt.statusCode, code: tt.code}}) + require.NoError(t, err) + + err = rs.DeleteBucket("container") + require.Equal(t, tt.wantErr, err != nil, "DeleteBucket: %v", err) + }) + } +} diff --git a/weed/remote_storage/gcs/gcs_storage_client.go b/weed/remote_storage/gcs/gcs_storage_client.go index 20de17a645..4d8d75b15b 100644 --- a/weed/remote_storage/gcs/gcs_storage_client.go +++ b/weed/remote_storage/gcs/gcs_storage_client.go @@ -488,6 +488,10 @@ func (gcs *gcsRemoteStorageClient) CreateBucket(name string) (err error) { func (gcs *gcsRemoteStorageClient) DeleteBucket(name string) (err error) { err = gcs.client.Bucket(name).Delete(context.Background()) if err != nil { + var gerr *googleapi.Error + if errors.Is(err, storage.ErrBucketNotExist) || (errors.As(err, &gerr) && gerr.Code == http.StatusNotFound) { + return nil + } return fmt.Errorf("delete bucket %s: %v", name, err) } return diff --git a/weed/remote_storage/gcs/gcs_storage_client_test.go b/weed/remote_storage/gcs/gcs_storage_client_test.go index 7ca67dac3c..5fdcff7d44 100644 --- a/weed/remote_storage/gcs/gcs_storage_client_test.go +++ b/weed/remote_storage/gcs/gcs_storage_client_test.go @@ -107,3 +107,22 @@ func TestGCSReadNotFoundClassification(t *testing.T) { }) } } + +func TestGCSDeleteBucketOfAMissingBucketSucceeds(t *testing.T) { + tests := []struct { + name string + status int + wantErr bool + }{ + {"missing bucket", http.StatusNotFound, false}, + {"bucket not empty", http.StatusConflict, true}, + {"access denied", http.StatusForbidden, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client := (&fakeGCS{readStatus: tt.status}).client(t) + err := client.DeleteBucket("bucket") + require.Equal(t, tt.wantErr, err != nil, "DeleteBucket: %v", err) + }) + } +} diff --git a/weed/remote_storage/remote_storage.go b/weed/remote_storage/remote_storage.go index 1bf2be90f4..c2091e4768 100644 --- a/weed/remote_storage/remote_storage.go +++ b/weed/remote_storage/remote_storage.go @@ -119,6 +119,7 @@ type RemoteStorageClient interface { DeleteFile(loc *remote_pb.RemoteStorageLocation) (err error) ListBuckets() ([]*Bucket, error) CreateBucket(name string) (err error) + // DeleteBucket succeeds when the bucket is already gone. DeleteBucket(name string) (err error) } diff --git a/weed/remote_storage/s3/s3_storage_client.go b/weed/remote_storage/s3/s3_storage_client.go index df9e02eddf..cfc4bea700 100644 --- a/weed/remote_storage/s3/s3_storage_client.go +++ b/weed/remote_storage/s3/s3_storage_client.go @@ -1001,6 +1001,10 @@ func (s *s3RemoteStorageClient) DeleteBucket(name string) (err error) { Bucket: aws.String(name), }) if err != nil { + var aerr awserr.Error + if errors.As(err, &aerr) && aerr.Code() == s3.ErrCodeNoSuchBucket { + return nil + } return fmt.Errorf("delete bucket %s: %v", name, err) } return diff --git a/weed/remote_storage/s3/s3_storage_client_test.go b/weed/remote_storage/s3/s3_storage_client_test.go index 06c1eb2b3c..a8de0e65a2 100644 --- a/weed/remote_storage/s3/s3_storage_client_test.go +++ b/weed/remote_storage/s3/s3_storage_client_test.go @@ -582,3 +582,23 @@ func TestS3MakeResolvesEnvCredentialsWhenAnonymousDefault(t *testing.T) { require.Equal(t, "env-access", value.AccessKeyID) require.Equal(t, "env-secret", value.SecretAccessKey) } + +func TestS3DeleteBucketOfAMissingBucketSucceeds(t *testing.T) { + tests := []struct { + name string + statusCode int + code string + wantErr bool + }{ + {"missing bucket", http.StatusNotFound, "NoSuchBucket", false}, + {"bucket not empty", http.StatusConflict, "BucketNotEmpty", true}, + {"access denied", http.StatusForbidden, "AccessDenied", true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client := newS3ClientWithTransport(t, &errorRoundTripper{statusCode: tt.statusCode, code: tt.code}, false) + err := client.DeleteBucket("bucket") + require.Equal(t, tt.wantErr, err != nil, "DeleteBucket: %v", err) + }) + } +}