From b335b9336b95bd87ebb6161a825e33e81989f8a6 Mon Sep 17 00:00:00 2001 From: dani <29378233+kheina@users.noreply.github.com> Date: Fri, 15 Mar 2024 13:33:47 -0400 Subject: [PATCH] implement PutObject functionality and associated tests (#10) --- plugin/service/storage/plugin.go | 89 +++++++- plugin/service/storage/plugin_test.go | 284 +++++++++++++++++++++++++- 2 files changed, 371 insertions(+), 2 deletions(-) diff --git a/plugin/service/storage/plugin.go b/plugin/service/storage/plugin.go index 37fb1b5..d982151 100644 --- a/plugin/service/storage/plugin.go +++ b/plugin/service/storage/plugin.go @@ -6,7 +6,11 @@ package storage import ( "bytes" "context" + "crypto/sha256" + "encoding/base64" "fmt" + "io" + "os" "path" "github.com/google/uuid" @@ -248,7 +252,90 @@ func (sp *StoragePlugin) GetObject(req *pb.GetObjectRequest, objServer pb.Storag // PutObject is a hook that reads a file stored on local disk and stores it to // an external object store. func (sp *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest) (*pb.PutObjectResponse, error) { - return nil, fmt.Errorf("unimplemented") + bucket := req.GetBucket() + if bucket == nil { + return nil, status.Error(codes.InvalidArgument, "no storage bucket information found") + } + if bucket.GetBucketName() == "" { + return nil, status.Error(codes.InvalidArgument, "storage bucket name is required") + } + if req.GetKey() == "" { + return nil, status.Error(codes.InvalidArgument, "object key is required") + } + if req.GetPath() == "" { + return nil, status.Error(codes.InvalidArgument, "path is required") + } + + sa, err := getStorageAttributes(bucket.GetAttributes()) + if err != nil { + return nil, err + } + + sec, err := getStorageSecrets(bucket.GetSecrets()) + if err != nil { + return nil, err + } + + file, err := os.Open(req.GetPath()) + if err != nil { + return nil, status.Errorf(codes.Internal, "failed to open file: %v", err) + } + defer file.Close() + + info, err := file.Stat() + if err != nil { + return nil, status.Errorf(codes.Internal, "failed to read file info: %v", err) + } + if info.IsDir() { + return nil, status.Error(codes.InvalidArgument, "path is not a file") + } + if info.Size() == 0 { + return nil, status.Error(codes.InvalidArgument, "file is empty") + } + + 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 nil, status.Errorf(codes.Internal, "failed to create minio sdk client: %v", err) + } + + hash := sha256.New() + if _, err := io.Copy(hash, file); err != nil { + return nil, status.Errorf(codes.Internal, "failed to calculate hash: %v", err) + } + checksum := base64.StdEncoding.EncodeToString(hash.Sum(nil)) + + if _, err = file.Seek(0, io.SeekStart); err != nil { + return nil, status.Errorf(codes.Internal, "failed to rewind file pointer: %v", err) + } + + key := path.Join(bucket.GetBucketPrefix(), req.GetKey()) + res, err := cl.PutObject(ctx, bucket.GetBucketName(), key, file, info.Size(), minio.PutObjectOptions{ + UserMetadata: map[string]string{ + "x-amz-checksum-algorithm": "SHA256", + "x-amz-checksum-sha256": checksum, + }, + }) + if err != nil { + return nil, status.Errorf(codes.Internal, "failed to put object into minio: %v", err) + } + + if res.ChecksumSHA256 == "" { + return nil, status.Error(codes.Internal, "missing checksum response from minio") + } + if checksum != res.ChecksumSHA256 { + return nil, status.Error(codes.Internal, "mismatched checksum") + } + decodedChecksum, err := base64.StdEncoding.DecodeString(res.ChecksumSHA256) + if err != nil { + return nil, status.Errorf(codes.Internal, "failed to decode checksum value from minio: %v", err) + } + return &pb.PutObjectResponse{ + ChecksumSha_256: decodedChecksum, + }, nil } // DeleteObjects deletes one or many files in an external object store via a diff --git a/plugin/service/storage/plugin_test.go b/plugin/service/storage/plugin_test.go index 43ac7f5..d9201e2 100644 --- a/plugin/service/storage/plugin_test.go +++ b/plugin/service/storage/plugin_test.go @@ -4,9 +4,12 @@ package storage import ( + "bufio" "bytes" "context" + "crypto/sha256" "fmt" + "os" "path" "testing" @@ -18,6 +21,7 @@ import ( "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/codes" "google.golang.org/grpc/status" @@ -991,7 +995,285 @@ func TestHeadObject(t *testing.T) { func TestGetObject(t *testing.T) {} -func TestPutObject(t *testing.T) {} +func TestPutObject(t *testing.T) { + td := t.TempDir() + + dirPath := func(p string) string { + return path.Join(td, p) + } + + ctx := context.Background() + server := internaltest.NewMinioServer(t) + + bucketName := "test-bucket" + require.NoError(t, server.Client.MakeBucket(ctx, bucketName, minio.MakeBucketOptions{})) + + cases := []struct { + name string + req *plugin.PutObjectRequest + setup func(*plugin.PutObjectRequest) (string, error) + err string + errCode codes.Code + }{ + { + name: "nilRequest", + err: "no storage bucket information found", + errCode: codes.InvalidArgument, + }, + { + name: "nilBucket", + req: &plugin.PutObjectRequest{Bucket: nil}, + err: "no storage bucket information found", + errCode: codes.InvalidArgument, + }, + { + name: "emptyBucket", + req: &plugin.PutObjectRequest{ + Bucket: &storagebuckets.StorageBucket{}, + }, + err: "storage bucket name is required", + errCode: codes.InvalidArgument, + }, + { + name: "emptyObjectKey", + req: &plugin.PutObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: bucketName, + }, + Key: "", + }, + err: "object key is required", + errCode: codes.InvalidArgument, + }, + { + name: "emptyFilePath", + req: &plugin.PutObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: bucketName, + }, + Key: "test-key", + Path: "", + }, + err: "path is required", + errCode: codes.InvalidArgument, + }, + { + name: "badStorageAttributes", + req: &plugin.PutObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: bucketName, + Attributes: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + "endpoint_url": structpb.NewStringValue("foo"), + }, + }, + }, + Key: "object-key", + Path: "file-path", + }, + err: "attributes.endpoint_url.format: unknown protocol, should be http:// or https://", + errCode: codes.InvalidArgument, + }, + { + name: "badStorageSecrets", + req: &plugin.PutObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: bucketName, + Attributes: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + "endpoint_url": structpb.NewStringValue("http://foo"), + }, + }, + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + "access_key_id": structpb.NewStringValue("foo"), + }, + }, + }, + Key: "object-key", + Path: "file-path", + }, + err: "secrets.secret_access_key: missing required value \"secret_access_key\"", + errCode: codes.InvalidArgument, + }, + { + name: "emptyFile", + req: &plugin.PutObjectRequest{ + 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: "test-key", + Path: dirPath("empty-test-file"), + }, + setup: func(req *plugin.PutObjectRequest) (string, error) { + file, err := os.Create(req.Path) + if err != nil { + return "", err + } + file.Close() + return "", nil + }, + err: "file is empty", + errCode: codes.InvalidArgument, + }, + { + name: "pathIsDir", + req: &plugin.PutObjectRequest{ + 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: "test-key", + Path: td, + }, + err: "path is not a file", + errCode: codes.InvalidArgument, + }, + { + name: "success", + req: &plugin.PutObjectRequest{ + 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: "test-key", + Path: dirPath("test-file"), + }, + setup: func(req *plugin.PutObjectRequest) (string, error) { + file, err := os.Create(req.Path) + if err != nil { + return "", err + } + defer file.Close() + + data := "test file data" + + if _, err = file.WriteString(data); err != nil { + return "", err + } + return data, nil + }, + }, + { + name: "successWithBucketPrefix", + req: &plugin.PutObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: bucketName, + BucketPrefix: "test-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: "test-key-with-prefix", + Path: dirPath("test-file-with-prefix"), + }, + setup: func(req *plugin.PutObjectRequest) (string, error) { + file, err := os.Create(req.Path) + if err != nil { + return "", err + } + defer file.Close() + + data := "test file data for bucket with prefix" + + if _, err = file.WriteString(data); err != nil { + return "", err + } + return data, nil + }, + }, + } + + for _, tt := range cases { + tt := tt + t.Run(tt.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + var checksum []byte + var content string + if tt.setup != nil { + var err error + content, err = tt.setup(tt.req) + require.NoError(err) + hash := sha256.New() + _, err = hash.Write([]byte(content)) + require.NoError(err) + checksum = hash.Sum(nil) + require.NoError(err) + } + + sp := new(StoragePlugin) + res, err := sp.PutObject(ctx, tt.req) + + if tt.err != "" { + assert.ErrorContains(err, tt.err) + assert.Equal(tt.errCode.String(), status.Code(err).String()) + return + } + + require.NotNil(res) + assert.NotEmpty(res.GetChecksumSha_256()) + assert.Equal(checksum, res.ChecksumSha_256) + + // check the actual file content is the same + cl, err := minio.New(server.ApiAddr, &minio.Options{ + Creds: credentials.NewStaticV4(server.RootUsername, server.RootPassword, ""), + Secure: false, + }) + require.NoError(err) + obj, err := cl.GetObject(ctx, bucketName, path.Join(tt.req.Bucket.GetBucketPrefix(), tt.req.GetKey()), minio.GetObjectOptions{}) + require.NoError(err) + defer obj.Close() + + contentLen := len([]byte(content)) + reader := bufio.NewReader(obj) + buffer := make([]byte, contentLen) + n, err := reader.Read(buffer) + require.NoError(err) + + assert.Equal(contentLen, n) + assert.Equal(content, string(buffer)) + }) + } +} func TestDeleteObjects(t *testing.T) {}