diff --git a/plugin/service/storage/plugin.go b/plugin/service/storage/plugin.go index d982151..f36a238 100644 --- a/plugin/service/storage/plugin.go +++ b/plugin/service/storage/plugin.go @@ -4,6 +4,7 @@ package storage import ( + "bufio" "bytes" "context" "crypto/sha256" @@ -27,6 +28,10 @@ import ( var _ pb.StoragePluginServiceServer = (*StoragePlugin)(nil) +const ( + defaultGetObjectChunkSize = uint32(65536) +) + // StoragePlugin implements the StoragePluginServiceServer interface for the // MinIO storage service plugin. type StoragePlugin struct { @@ -246,7 +251,71 @@ func (sp *StoragePlugin) HeadObject(ctx context.Context, req *pb.HeadObjectReque // GetObject is a hook that retrieves objects. func (sp *StoragePlugin) GetObject(req *pb.GetObjectRequest, objServer pb.StoragePluginService_GetObjectServer) error { - return fmt.Errorf("unimplemented") + bucket := req.GetBucket() + if bucket == nil { + return status.Error(codes.InvalidArgument, "no storage bucket information found") + } + if bucket.GetBucketName() == "" { + return status.Error(codes.InvalidArgument, "storage bucket name is required") + } + + if req.GetKey() == "" { + return status.Error(codes.InvalidArgument, "object key is required") + } + + sa, err := getStorageAttributes(bucket.GetAttributes()) + if err != nil { + return err + } + + sec, err := getStorageSecrets(bucket.GetSecrets()) + if err != nil { + return err + } + + cl, err := minio.New(sa.EndpointUrl, &minio.Options{ + Creds: credentials.NewStaticV4(sec.AccessKeyId, sec.SecretAccessKey, ""), + Secure: sa.UseSSL, + Region: sa.Region, + }) + if err != nil { + return status.Errorf(codes.Unknown, "failed to create minio sdk client: %v", err) + } + + objKey := path.Join(bucket.GetBucketPrefix(), req.GetKey()) + obj, err := cl.GetObject(objServer.Context(), bucket.GetBucketName(), objKey, minio.GetObjectOptions{}) + if err != nil { + return status.Errorf(codes.Unknown, "failed to get object: %v", err) + } + defer obj.Close() + + chunkSize := req.GetChunkSize() + if chunkSize == 0 { + chunkSize = defaultGetObjectChunkSize + } + + rd := bufio.NewReader(obj) + for { + select { + case <-objServer.Context().Done(): + return status.Errorf(codes.Canceled, "server context done while streaming: %v", objServer.Context().Err()) + default: + buffer := make([]byte, chunkSize) + n, err := rd.Read(buffer) + if err != nil && err != io.EOF { + return status.Errorf(codes.Unknown, "failed to read chunk from minio object: %v", err) + } + if n > 0 { + err := objServer.Send(&pb.GetObjectResponse{FileChunk: buffer[:n]}) + if err != nil { + return status.Errorf(codes.Unknown, "failed to send chunk to client: %v", err) + } + } + if err == io.EOF { + return nil + } + } + } } // PutObject is a hook that reads a file stored on local disk and stores it to diff --git a/plugin/service/storage/plugin_test.go b/plugin/service/storage/plugin_test.go index d9201e2..973198f 100644 --- a/plugin/service/storage/plugin_test.go +++ b/plugin/service/storage/plugin_test.go @@ -9,6 +9,7 @@ import ( "context" "crypto/sha256" "fmt" + "io" "os" "path" "testing" @@ -18,11 +19,13 @@ import ( internaltest "github.com/hashicorp/boundary-plugin-minio/internal/testing" "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/storagebuckets" "github.com/hashicorp/boundary/sdk/pbs/plugin" + pb "github.com/hashicorp/boundary/sdk/pbs/plugin" "github.com/minio/madmin-go/v3" "github.com/minio/minio-go/v7" "github.com/minio/minio-go/v7/pkg/credentials" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/testing/protocmp" @@ -125,6 +128,57 @@ var ( }`) ) +type getObjectServerMock struct { + grpc.ServerStream + ctx context.Context + sendErr error + clientCtxDone bool + chunks []*pb.GetObjectResponse +} + +// Context returns the object server's context. +func (s *getObjectServerMock) Context() context.Context { + if s.ctx == nil { + return context.Background() + } + return s.ctx +} + +// Send streams a response message to the client. +func (g *getObjectServerMock) Send(rsp *pb.GetObjectResponse) error { + if g.sendErr != nil { + return g.sendErr + } + if rsp == nil { + return fmt.Errorf("getobject stream mock: nil response") + } + if g.chunks == nil { + g.chunks = []*pb.GetObjectResponse{} + } + g.chunks = append(g.chunks, rsp) + if g.clientCtxDone { + ctx, cancel := context.WithCancel(g.ctx) + g.ctx = ctx + cancel() + } + return nil +} + +// Recv streams a response message from the server. +func (g *getObjectServerMock) Recv() (*pb.GetObjectResponse, error) { + if g.chunks == nil { + return nil, fmt.Errorf("getobject stream mock: nil chunks") + } + if len(g.chunks) == 0 { + return nil, io.EOF + } + + // Pop a chunk off and return it. + chunk := g.chunks[0] + g.chunks = g.chunks[1:] + return chunk, nil +} + func TestOnCreateStorageBucket(t *testing.T) { ctx := context.Background() server := internaltest.NewMinioServer(t) @@ -993,7 +1047,398 @@ func TestHeadObject(t *testing.T) { } } -func TestGetObject(t *testing.T) {} +func TestGetObject(t *testing.T) { + ctx := context.Background() + server := internaltest.NewMinioServer(t) + + bucketName := "test-bucket" + err := server.Client.MakeBucket(ctx, bucketName, minio.MakeBucketOptions{}) + require.NoError(t, err) + + tests := []struct { + name string + in func(t *testing.T) (*plugin.GetObjectRequest, []byte) + mockSendErr error + mockCtxDone bool + expErrMsg string + expErrCode codes.Code + }{ + { + name: "nilBucket", + in: func(t *testing.T) (*pb.GetObjectRequest, []byte) { + return &pb.GetObjectRequest{Bucket: nil}, []byte{} + }, + expErrMsg: "no storage bucket information found", + expErrCode: codes.InvalidArgument, + }, + { + name: "noBucketName", + in: func(t *testing.T) (*pb.GetObjectRequest, []byte) { + return &pb.GetObjectRequest{ + Bucket: &storagebuckets.StorageBucket{BucketName: ""}, + }, []byte{} + }, + expErrMsg: "storage bucket name is required", + expErrCode: codes.InvalidArgument, + }, + { + name: "emptyObjectKey", + in: func(t *testing.T) (*pb.GetObjectRequest, []byte) { + return &pb.GetObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: bucketName, + }, + Key: "", + }, []byte{} + }, + expErrMsg: "object key is required", + expErrCode: codes.InvalidArgument, + }, + { + name: "badStorageAttributes", + in: func(t *testing.T) (*pb.GetObjectRequest, []byte) { + return &pb.GetObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: bucketName, + }, + Key: "object-name", + }, []byte{} + }, + expErrMsg: "empty attributes input", + expErrCode: codes.InvalidArgument, + }, + { + name: "badStorageSecrets", + in: func(t *testing.T) (*pb.GetObjectRequest, []byte) { + return &pb.GetObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: bucketName, + Attributes: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstEndpointUrl: structpb.NewStringValue("http://foo"), + }, + }, + }, + Key: "object-name", + }, []byte{} + }, + expErrMsg: "empty secrets input", + expErrCode: codes.InvalidArgument, + }, + { + name: "getObjectErr", + in: func(t *testing.T) (*pb.GetObjectRequest, []byte) { + return &pb.GetObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: "invalid..bucket..name", + Attributes: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstEndpointUrl: structpb.NewStringValue("http://localhost:1"), + }, + }, + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstAccessKeyId: structpb.NewStringValue("foo"), + ConstSecretAccessKey: structpb.NewStringValue("bar"), + }, + }, + }, + Key: "object-name", + }, []byte{} + }, + expErrMsg: "failed to get object: Bucket name contains invalid characters", + expErrCode: codes.Unknown, + }, + { + name: "readObjectErr", + in: func(t *testing.T) (*pb.GetObjectRequest, []byte) { + return &pb.GetObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: bucketName, + Attributes: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstEndpointUrl: structpb.NewStringValue("http://localhost:1"), + }, + }, + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstAccessKeyId: structpb.NewStringValue("foo"), + ConstSecretAccessKey: structpb.NewStringValue("bar"), + }, + }, + }, + Key: "object-name", + }, []byte{} + }, + expErrMsg: "failed to read chunk from minio object", + expErrCode: codes.Unknown, + }, + { + name: "objectDoesntExist", + in: func(t *testing.T) (*pb.GetObjectRequest, []byte) { + return &pb.GetObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: bucketName, + Attributes: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr), + }, + }, + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstAccessKeyId: structpb.NewStringValue(server.ServiceAccountAccessKeyId), + ConstSecretAccessKey: structpb.NewStringValue(server.ServiceAccountSecretAccessKey), + }, + }, + }, + Key: "object-name", + }, []byte{} + }, + expErrMsg: "failed to read chunk from minio object: The specified key does not exist.", + expErrCode: codes.Unknown, + }, + { + name: "objServerSendErr", + in: func(t *testing.T) (*pb.GetObjectRequest, []byte) { + objName := uuid.New().String() + rd := bytes.NewReader([]byte("test")) + _, err := server.Client.PutObject(ctx, bucketName, objName, rd, rd.Size(), minio.PutObjectOptions{}) + require.NoError(t, err) + + return &pb.GetObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: bucketName, + Attributes: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr), + }, + }, + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstAccessKeyId: structpb.NewStringValue(server.ServiceAccountAccessKeyId), + ConstSecretAccessKey: structpb.NewStringValue(server.ServiceAccountSecretAccessKey), + }, + }, + }, + Key: objName, + }, []byte{} + }, + mockSendErr: fmt.Errorf("oops error"), + expErrMsg: "failed to send chunk to client: oops error", + expErrCode: codes.Unknown, + }, + { + name: "contextCancellation", + in: func(t *testing.T) (*pb.GetObjectRequest, []byte) { + objName := uuid.New().String() + + data := []byte{} + for i := 0; i < 8192; i += len(objName) { + data = append(data, []byte(objName)...) + } + rd := bytes.NewReader(data) + _, err := server.Client.PutObject(ctx, bucketName, objName, rd, rd.Size(), minio.PutObjectOptions{}) + require.NoError(t, err) + + return &pb.GetObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: bucketName, + Attributes: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr), + }, + }, + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstAccessKeyId: structpb.NewStringValue(server.ServiceAccountAccessKeyId), + ConstSecretAccessKey: structpb.NewStringValue(server.ServiceAccountSecretAccessKey), + }, + }, + }, + Key: objName, + ChunkSize: 2048, + }, data + }, + mockCtxDone: true, + expErrMsg: "server context done while streaming: context canceled", + expErrCode: codes.Canceled, + }, + { + name: "successSingleChunk", + in: func(t *testing.T) (*pb.GetObjectRequest, []byte) { + objName := uuid.New().String() + + data := []byte{} + for i := 0; i < 8192; i += len(objName) { + data = append(data, []byte(objName)...) + } + rd := bytes.NewReader(data) + _, err := server.Client.PutObject(ctx, bucketName, objName, rd, rd.Size(), minio.PutObjectOptions{}) + require.NoError(t, err) + + return &pb.GetObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: bucketName, + Attributes: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr), + }, + }, + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstAccessKeyId: structpb.NewStringValue(server.ServiceAccountAccessKeyId), + ConstSecretAccessKey: structpb.NewStringValue(server.ServiceAccountSecretAccessKey), + }, + }, + }, + Key: objName, + ChunkSize: 16384, + }, data + }, + }, + { + name: "successCustomPrefix", + in: func(t *testing.T) (*pb.GetObjectRequest, []byte) { + objName := uuid.New().String() + + data := []byte{} + for i := 0; i < 8192; i += len(objName) { + data = append(data, []byte(objName)...) + } + rd := bytes.NewReader(data) + + prefix := "prefix" + _, err := server.Client.PutObject(ctx, bucketName, path.Join(prefix, objName), rd, rd.Size(), minio.PutObjectOptions{}) + require.NoError(t, err) + + return &pb.GetObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: bucketName, + BucketPrefix: prefix, + Attributes: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr), + }, + }, + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstAccessKeyId: structpb.NewStringValue(server.ServiceAccountAccessKeyId), + ConstSecretAccessKey: structpb.NewStringValue(server.ServiceAccountSecretAccessKey), + }, + }, + }, + Key: objName, + }, data + }, + }, + { + name: "successMultipleChunksDefaultChunkSize", + in: func(t *testing.T) (*pb.GetObjectRequest, []byte) { + objName := uuid.New().String() + + data := []byte{} + for i := 0; i < 102_400; i += len(objName) { + data = append(data, []byte(objName)...) + } + rd := bytes.NewReader(data) + _, err := server.Client.PutObject(ctx, bucketName, objName, rd, rd.Size(), minio.PutObjectOptions{}) + require.NoError(t, err) + + return &pb.GetObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: bucketName, + Attributes: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr), + }, + }, + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstAccessKeyId: structpb.NewStringValue(server.ServiceAccountAccessKeyId), + ConstSecretAccessKey: structpb.NewStringValue(server.ServiceAccountSecretAccessKey), + }, + }, + }, + Key: objName, + }, data + }, + }, + { + name: "successMultipleChunksCustomChunkSize", + in: func(t *testing.T) (*pb.GetObjectRequest, []byte) { + objName := uuid.New().String() + + data := []byte{} + for i := 0; i < 10_420; i += len(objName) { + data = append(data, []byte(objName)...) + } + rd := bytes.NewReader(data) + _, err := server.Client.PutObject(ctx, bucketName, objName, rd, rd.Size(), minio.PutObjectOptions{}) + require.NoError(t, err) + + return &pb.GetObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: bucketName, + Attributes: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr), + }, + }, + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + ConstAccessKeyId: structpb.NewStringValue(server.ServiceAccountAccessKeyId), + ConstSecretAccessKey: structpb.NewStringValue(server.ServiceAccountSecretAccessKey), + }, + }, + }, + Key: objName, + ChunkSize: 4096, + }, data + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + objServer := &getObjectServerMock{ + ctx: ctx, + clientCtxDone: tt.mockCtxDone, + sendErr: tt.mockSendErr, + } + + req, expFileBytes := tt.in(t) + sp := new(StoragePlugin) + err := sp.GetObject(req, objServer) + if tt.expErrMsg != "" { + require.ErrorContains(t, err, tt.expErrMsg) + require.Equal(t, tt.expErrCode.String(), status.Code(err).String()) + return + } + require.NoError(t, err) + + expChunkSize := defaultGetObjectChunkSize + if req.ChunkSize > 0 { + expChunkSize = req.ChunkSize + } + + bb := bytes.NewBuffer([]byte{}) + for { + rsp, err := objServer.Recv() + if err == io.EOF { + break + } + require.NoError(t, err) + require.LessOrEqual(t, uint32(len(rsp.GetFileChunk())), expChunkSize) + + n, err := bb.Write(rsp.GetFileChunk()) + require.NoError(t, err) + require.EqualValues(t, len(rsp.GetFileChunk()), n) + } + + require.Equal(t, expFileBytes, bb.Bytes()) + }) + } +} func TestPutObject(t *testing.T) { td := t.TempDir()