add delete objects command to proto and loopback (#785)

This commit is contained in:
dani authored and Louis Ruch committed 2024-01-16 13:53:38 -08:00
1 parent e3971decf4
commit c43ded06a6
8 files changed
+592 -82

No files matched your search

+4
View File
@@ -102,3 +102,7 @@ func (tpc *WrappingPluginStorageClient) GetObject(ctx context.Context, req *plgp
func (tpc *WrappingPluginStorageClient) PutObject(ctx context.Context, req *plgpb.PutObjectRequest, opts ...grpc.CallOption) (*plgpb.PutObjectResponse, error) {
return tpc.Server.PutObject(ctx, req)
}
func (tpc *WrappingPluginStorageClient) DeleteObjects(ctx context.Context, req *plgpb.DeleteObjectsRequest, opts ...grpc.CallOption) (*plgpb.DeleteObjectsResponse, error) {
return tpc.Server.DeleteObjects(ctx, req)
}
+1
View File
@@ -75,6 +75,7 @@ func NewLoopbackPlugin(opt ...TestOption) (*LoopbackPlugin, error) {
ret.HeadObjectFn = ret.headObject
ret.GetObjectFn = ret.getObject
ret.PutObjectFn = ret.putObject
ret.DeleteObjectsFn = ret.deleteObjects
if len(opts.withMockBuckets) > 0 {
ret.buckets = opts.withMockBuckets
}
+3
View File
@@ -36,6 +36,9 @@ const (
// Will return an error for the PutObject method
PutObject
// Will return an error for the DeleteObjects method
DeleteObjects
)
const (
+55
View File
@@ -10,6 +10,7 @@ import (
"io"
"os"
"path"
"strings"
"sync"
"time"
@@ -32,6 +33,7 @@ type TestPluginStorageServer struct {
HeadObjectFn func(context.Context, *plgpb.HeadObjectRequest) (*plgpb.HeadObjectResponse, error)
GetObjectFn func(*plgpb.GetObjectRequest, plgpb.StoragePluginService_GetObjectServer) error
PutObjectFn func(context.Context, *plgpb.PutObjectRequest) (*plgpb.PutObjectResponse, error)
DeleteObjectsFn func(context.Context, *plgpb.DeleteObjectsRequest) (*plgpb.DeleteObjectsResponse, error)
plgpb.UnimplementedStoragePluginServiceServer
}
@@ -84,6 +86,13 @@ func (t TestPluginStorageServer) PutObject(ctx context.Context, req *plgpb.PutOb
return t.PutObjectFn(ctx, req)
}
func (t TestPluginStorageServer) DeleteObjects(ctx context.Context, req *plgpb.DeleteObjectsRequest) (*plgpb.DeleteObjectsResponse, error) {
if t.DeleteObjectsFn == nil {
return t.UnimplementedStoragePluginServiceServer.DeleteObjects(ctx, req)
}
return t.DeleteObjectsFn(ctx, req)
}
type (
BucketName string
ObjectName string
@@ -406,6 +415,52 @@ func (l *LoopbackStorage) putObject(ctx context.Context, req *plgpb.PutObjectReq
}, nil
}
func (l *LoopbackStorage) deleteObjects(ctx context.Context, req *plgpb.DeleteObjectsRequest) (*plgpb.DeleteObjectsResponse, error) {
const op = "loopback.(LoopbackStorage).deleteObjects"
if req == nil {
return nil, status.Errorf(codes.InvalidArgument, "%s: request is nil", op)
}
if req.GetBucket() == nil {
return nil, status.Errorf(codes.InvalidArgument, "%s: missing storage bucket", op)
}
if req.GetBucket().GetAttributes() == nil {
return nil, status.Errorf(codes.InvalidArgument, "%s: missing attributes", op)
}
if req.GetKeyPrefix() == "" {
return nil, status.Errorf(codes.InvalidArgument, "%s; missing key prefix", op)
}
l.m.Lock()
defer l.m.Unlock()
bucket, ok := l.buckets[BucketName(req.GetBucket().GetBucketName())]
if !ok {
return nil, status.Errorf(codes.NotFound, "%s: bucket not found", op)
}
for _, err := range l.errs {
if err.match(req.GetBucket(), req.GetKeyPrefix(), DeleteObjects) {
return nil, status.Errorf(err.ErrCode, "%s: %s", op, err.ErrMsg)
}
}
prefix := ObjectName(req.GetBucket().GetBucketPrefix() + req.GetKeyPrefix())
var deleted uint32 = 0
if req.GetRecursive() {
for key := range bucket {
if strings.HasPrefix(string(key), string(prefix)) {
delete(bucket, key)
deleted++
}
}
} else {
_, ok := bucket[prefix]
if ok {
delete(bucket, prefix)
deleted++
}
}
return &plgpb.DeleteObjectsResponse{
ObjectsDeleted: deleted,
}, nil
}
// CloneBucket returns a clone of the bucket.
// returns nil when the bucket is not found.
func (l *LoopbackStorage) CloneBucket(name string) Bucket {
+195
View File
@@ -7,6 +7,7 @@ import (
"bytes"
"context"
"crypto/sha256"
"fmt"
"io"
"io/fs"
"os"
@@ -14,6 +15,7 @@ import (
"strings"
"sync"
"testing"
"time"
"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/storagebuckets"
plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin"
@@ -1224,6 +1226,199 @@ func TestLoopbackPutObject(t *testing.T) {
}
}
func TestLoopbackDeleteObjects(t *testing.T) {
require := tr.New(t)
now := time.Now()
var zero int64 = 0
mockStorageMapData := map[BucketName]Bucket{
"object_store": {
ObjectName("test-file-1"): &storagePluginStorageInfo{
lastModified: &now,
contentLength: &zero,
DataChunks: []Chunk{},
},
ObjectName("test-file-2"): &storagePluginStorageInfo{
lastModified: &now,
contentLength: &zero,
DataChunks: []Chunk{},
},
ObjectName("test-file-3"): &storagePluginStorageInfo{
lastModified: &now,
contentLength: &zero,
DataChunks: []Chunk{},
},
ObjectName("abc/file-1"): &storagePluginStorageInfo{
lastModified: &now,
contentLength: &zero,
DataChunks: []Chunk{},
},
ObjectName("abc/file-2"): &storagePluginStorageInfo{
lastModified: &now,
contentLength: &zero,
DataChunks: []Chunk{},
},
ObjectName("abc/file-3"): &storagePluginStorageInfo{
lastModified: &now,
contentLength: &zero,
DataChunks: []Chunk{},
},
},
"object_store_err": {},
}
plg, err := NewLoopbackPlugin(
WithMockBuckets(mockStorageMapData),
WithMockError(PluginMockError{
BucketName: "object_store_err",
ObjectKey: "mock_object",
ErrMsg: "invalid credentials",
ErrCode: codes.PermissionDenied,
}),
)
require.NoError(err)
require.NotNil(plg)
client := NewWrappingPluginStorageClient(plg)
tests := []struct {
name string
request *plgpb.DeleteObjectsRequest
expected *plgpb.DeleteObjectsResponse
expectedErr string
check func() error
}{
{
name: "valid delete object no recursive",
request: &plgpb.DeleteObjectsRequest{
KeyPrefix: "test-file-1",
Bucket: &storagebuckets.StorageBucket{
BucketName: "object_store",
Attributes: &structpb.Struct{
Fields: map[string]*structpb.Value{
"endpoint": structpb.NewStringValue("0.0.0.0"),
},
},
},
Recursive: false,
},
expected: &plgpb.DeleteObjectsResponse{
ObjectsDeleted: 1,
},
check: func() error {
filename := "test-file-1"
bucket, _ := mockStorageMapData["object_store"]
_, ok := bucket[ObjectName(filename)]
if ok {
return fmt.Errorf("failed to delete object %s", filename)
}
return nil
},
}, {
name: "valid delete objects recursive",
request: &plgpb.DeleteObjectsRequest{
KeyPrefix: "abc/",
Bucket: &storagebuckets.StorageBucket{
BucketName: "object_store",
Attributes: &structpb.Struct{
Fields: map[string]*structpb.Value{
"endpoint": structpb.NewStringValue("0.0.0.0"),
},
},
},
Recursive: true,
},
expected: &plgpb.DeleteObjectsResponse{
ObjectsDeleted: 3,
},
check: func() error {
filenames := []string{
"abc/file-1",
"abc/file-2",
"abc/file-3",
}
bucket, _ := mockStorageMapData["object_store"]
for _, filename := range filenames {
_, ok := bucket[ObjectName(filename)]
if ok {
return fmt.Errorf("failed to delete object %s", filename)
}
}
return nil
},
}, {
name: "valid delete objects none",
request: &plgpb.DeleteObjectsRequest{
KeyPrefix: "abcd/",
Bucket: &storagebuckets.StorageBucket{
BucketName: "object_store",
Attributes: &structpb.Struct{
Fields: map[string]*structpb.Value{
"endpoint": structpb.NewStringValue("0.0.0.0"),
},
},
},
Recursive: true,
},
expected: &plgpb.DeleteObjectsResponse{
ObjectsDeleted: 0,
},
check: func() error {
return nil
},
}, {
name: "delete objects object doesn't exist",
request: &plgpb.DeleteObjectsRequest{
KeyPrefix: "non-existent-object",
Bucket: &storagebuckets.StorageBucket{
BucketName: "object_store",
Attributes: &structpb.Struct{
Fields: map[string]*structpb.Value{
"endpoint": structpb.NewStringValue("0.0.0.0"),
},
},
},
Recursive: false,
},
expected: &plgpb.DeleteObjectsResponse{
ObjectsDeleted: 0,
},
check: func() error {
return nil
},
}, {
name: "delete objects bucket doesn't exist",
request: &plgpb.DeleteObjectsRequest{
KeyPrefix: "abc/",
Bucket: &storagebuckets.StorageBucket{
BucketName: "non-existent-bucket",
Attributes: &structpb.Struct{
Fields: map[string]*structpb.Value{
"endpoint": structpb.NewStringValue("0.0.0.0"),
},
},
},
Recursive: false,
},
expectedErr: "loopback.(LoopbackStorage).deleteObjects: bucket not found",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
resp, err := client.DeleteObjects(context.Background(), tt.request)
if tt.expectedErr != "" {
require.Error(err)
require.Nil(resp)
require.ErrorContains(err, tt.expectedErr)
return
}
require.Equal(tt.expected, resp)
require.NoError(tt.check())
})
}
}
func TestLoopbackStoragePlugin(t *testing.T) {
td := t.TempDir()