diff --git a/internal/plugin/loopback/client.go b/internal/plugin/loopback/client.go index c0e0df354..033cb8477 100644 --- a/internal/plugin/loopback/client.go +++ b/internal/plugin/loopback/client.go @@ -99,10 +99,6 @@ func (tpc *WrappingPluginStorageClient) GetObject(ctx context.Context, req *plgp return stream.client, nil } -func (tpc *WrappingPluginStorageClient) PutObject(ctx context.Context, opts ...grpc.CallOption) (plgpb.StoragePluginService_PutObjectClient, error) { - stream := newPutObjectStream() - if err := tpc.Server.PutObject(stream.server); err != nil { - return nil, err - } - return stream.client, nil +func (tpc *WrappingPluginStorageClient) PutObject(ctx context.Context, req *plgpb.PutObjectRequest, opts ...grpc.CallOption) (*plgpb.PutObjectResponse, error) { + return tpc.Server.PutObject(ctx, req) } diff --git a/internal/plugin/loopback/loopback.go b/internal/plugin/loopback/loopback.go index c45821f99..1ee6a9bbb 100644 --- a/internal/plugin/loopback/loopback.go +++ b/internal/plugin/loopback/loopback.go @@ -52,8 +52,9 @@ func NewLoopbackPlugin(opt ...TestOption) (*LoopbackPlugin, error) { hostMap: make(map[string][]*loopbackPluginHostInfo), }, LoopbackStorage: &LoopbackStorage{ - buckets: map[BucketName]Bucket{"default": {}}, - errs: make([]PluginMockError, 0), + chunksSize: opts.withChunkSize, + buckets: map[BucketName]Bucket{"default": {}}, + errs: make([]PluginMockError, 0), }, } diff --git a/internal/plugin/loopback/options.go b/internal/plugin/loopback/options.go index c189a9ff7..119052b0d 100644 --- a/internal/plugin/loopback/options.go +++ b/internal/plugin/loopback/options.go @@ -35,6 +35,7 @@ type TestOption func(*TestOptions) error type TestOptions struct { withMockBuckets map[BucketName]Bucket withMockError []PluginMockError + withChunkSize int } // getTestOpts - iterate the inbound Options and return a struct @@ -49,7 +50,9 @@ func getTestOpts(opt ...TestOption) (TestOptions, error) { } func getDefaultTestOptions() TestOptions { - return TestOptions{} + return TestOptions{ + withChunkSize: 8, + } } // WithMockBuckets provides an option to create mocked external object store buckets. @@ -67,3 +70,11 @@ func WithMockError(err PluginMockError) TestOption { return nil } } + +// WithChunkSize provides an option to set the chunkSize used for grpc streams. +func WithChunkSize(size int) TestOption { + return func(o *TestOptions) error { + o.withChunkSize = size + return nil + } +} diff --git a/internal/plugin/loopback/storage.go b/internal/plugin/loopback/storage.go index 984ef923c..eb359d2d7 100644 --- a/internal/plugin/loopback/storage.go +++ b/internal/plugin/loopback/storage.go @@ -9,6 +9,7 @@ import ( "crypto/sha256" "fmt" "io" + "os" "strings" "sync" "time" @@ -29,7 +30,7 @@ type TestPluginStorageServer struct { ValidatePermissionsFn func(context.Context, *plgpb.ValidatePermissionsRequest) (*plgpb.ValidatePermissionsResponse, error) HeadObjectFn func(context.Context, *plgpb.HeadObjectRequest) (*plgpb.HeadObjectResponse, error) GetObjectFn func(*plgpb.GetObjectRequest, plgpb.StoragePluginService_GetObjectServer) error - PutObjectFn func(plgpb.StoragePluginService_PutObjectServer) error + PutObjectFn func(context.Context, *plgpb.PutObjectRequest) (*plgpb.PutObjectResponse, error) plgpb.UnimplementedStoragePluginServiceServer } @@ -75,11 +76,11 @@ func (t TestPluginStorageServer) GetObject(req *plgpb.GetObjectRequest, stream p return t.GetObjectFn(req, stream) } -func (t TestPluginStorageServer) PutObject(stream plgpb.StoragePluginService_PutObjectServer) error { +func (t TestPluginStorageServer) PutObject(ctx context.Context, req *plgpb.PutObjectRequest) (*plgpb.PutObjectResponse, error) { if t.PutObjectFn == nil { - return t.UnimplementedStoragePluginServiceServer.PutObject(stream) + return t.UnimplementedStoragePluginServiceServer.PutObject(ctx, req) } - return t.PutObjectFn(stream) + return t.PutObjectFn(ctx, req) } type ( @@ -106,19 +107,20 @@ type storagePluginStorageInfo struct { type LoopbackStorage struct { m sync.Mutex - buckets map[BucketName]Bucket - errs []PluginMockError + chunksSize int + buckets map[BucketName]Bucket + errs []PluginMockError } func (l *LoopbackStorage) onCreateStorageBucket(ctx context.Context, req *plgpb.OnCreateStorageBucketRequest) (*plgpb.OnCreateStorageBucketResponse, error) { - const op = "loopback.(LoopbackPlugin).onCreateStorageBucket" + const op = "loopback.(LoopbackStorage).onCreateStorageBucket" if req == nil { return nil, status.Errorf(codes.InvalidArgument, "%s: request is nil", op) } - if req.Bucket == nil { + if req.GetBucket() == nil { return nil, status.Errorf(codes.InvalidArgument, "%s: missing storage bucket", op) } - if req.GetBucket().Secrets == nil { + if req.GetBucket().GetSecrets() == nil { return nil, status.Errorf(codes.InvalidArgument, "%s: missing secrets", op) } l.m.Lock() @@ -139,14 +141,14 @@ func (l *LoopbackStorage) onCreateStorageBucket(ctx context.Context, req *plgpb. } func (l *LoopbackStorage) onUpdateStorageBucket(ctx context.Context, req *plgpb.OnUpdateStorageBucketRequest) (*plgpb.OnUpdateStorageBucketResponse, error) { - const op = "loopback.(LoopbackPlugin).onUpdateStorageBucket" + const op = "loopback.(LoopbackStorage).onUpdateStorageBucket" if req == nil { return nil, status.Errorf(codes.InvalidArgument, "%s: request is nil", op) } - if req.NewBucket == nil { + if req.GetNewBucket() == nil { return nil, status.Errorf(codes.InvalidArgument, "%s: missing storage bucket", op) } - if req.GetNewBucket().Secrets == nil { + if req.GetNewBucket().GetSecrets() == nil { return nil, status.Errorf(codes.InvalidArgument, "%s: missing secrets", op) } l.m.Lock() @@ -167,28 +169,28 @@ func (l *LoopbackStorage) onUpdateStorageBucket(ctx context.Context, req *plgpb. } func (l *LoopbackStorage) onDeleteStorageBucket(ctx context.Context, req *plgpb.OnDeleteStorageBucketRequest) (*plgpb.OnDeleteStorageBucketResponse, error) { - const op = "loopback.(LoopbackPlugin).onDeleteStorageBucket" + const op = "loopback.(LoopbackStorage).onDeleteStorageBucket" if req == nil { return nil, status.Errorf(codes.InvalidArgument, "%s: request is nil", op) } - if req.Bucket == nil { + if req.GetBucket() == nil { return nil, status.Errorf(codes.InvalidArgument, "%s: missing storage bucket", op) } - if req.GetBucket().Secrets == nil { + if req.GetBucket().GetSecrets() == nil { return nil, status.Errorf(codes.InvalidArgument, "%s: missing secrets", op) } return &plgpb.OnDeleteStorageBucketResponse{}, nil } func (l *LoopbackStorage) validatePermissions(ctx context.Context, req *plgpb.ValidatePermissionsRequest) (*plgpb.ValidatePermissionsResponse, error) { - const op = "loopback.(LoopbackPlugin).validatePermissions" + const op = "loopback.(LoopbackStorage).validatePermissions" if req == nil { return nil, status.Errorf(codes.InvalidArgument, "%s: request is nil", op) } - if req.Bucket == nil { + if req.GetBucket() == nil { return nil, status.Errorf(codes.InvalidArgument, "%s: missing storage bucket", op) } - if req.GetBucket().Secrets == nil { + if req.GetBucket().GetSecrets() == nil { return nil, status.Errorf(codes.InvalidArgument, "%s: missing secrets", op) } l.m.Lock() @@ -205,17 +207,17 @@ func (l *LoopbackStorage) validatePermissions(ctx context.Context, req *plgpb.Va } func (l *LoopbackStorage) headObject(ctx context.Context, req *plgpb.HeadObjectRequest) (*plgpb.HeadObjectResponse, error) { - const op = "loopback.(LoopbackPlugin).headObject" + const op = "loopback.(LoopbackStorage).headObject" if req == nil { return nil, status.Errorf(codes.InvalidArgument, "%s: request is nil", op) } - if req.Bucket == nil { + if req.GetBucket() == nil { return nil, status.Errorf(codes.InvalidArgument, "$%s: missing storage bucket", op) } - if req.GetBucket().Secrets == nil { + if req.GetBucket().GetSecrets() == nil { return nil, status.Errorf(codes.InvalidArgument, "%s: missing secrets", op) } - if req.Key == "" { + if req.GetKey() == "" { return nil, status.Errorf(codes.InvalidArgument, "%s; missing object key", op) } l.m.Lock() @@ -245,17 +247,17 @@ func (l *LoopbackStorage) headObject(ctx context.Context, req *plgpb.HeadObjectR } func (l *LoopbackStorage) getObject(req *plgpb.GetObjectRequest, stream plgpb.StoragePluginService_GetObjectServer) error { - const op = "loopback.(LoopbackPlugin).getObject" + const op = "loopback.(LoopbackStorage).getObject" if req == nil { return status.Errorf(codes.InvalidArgument, "%s: request is nil", op) } - if req.Bucket == nil { + if req.GetBucket() == nil { return status.Errorf(codes.InvalidArgument, "%s: missing storage bucket", op) } - if req.GetBucket().Secrets == nil { + if req.GetBucket().GetSecrets() == nil { return status.Errorf(codes.InvalidArgument, "%s: missing secrets", op) } - if req.Key == "" { + if req.GetKey() == "" { return status.Errorf(codes.InvalidArgument, "%s; missing object key", op) } l.m.Lock() @@ -288,110 +290,93 @@ func (l *LoopbackStorage) getObject(req *plgpb.GetObjectRequest, stream plgpb.St return nil } -func (l *LoopbackStorage) putObject(stream plgpb.StoragePluginService_PutObjectServer) error { - const op = "loopback.(LoopbackPlugin).putObject" - if stream == nil { - return status.Errorf(codes.Internal, "%s: missing stream", op) +func (l *LoopbackStorage) putObject(ctx context.Context, req *plgpb.PutObjectRequest) (*plgpb.PutObjectResponse, error) { + const op = "loopback.(LoopbackStorage).putObject" + if req == nil { + return nil, status.Errorf(codes.InvalidArgument, "%s: request is nil", op) } - go func() { - var ( - request *plgpb.PutObjectRequest_Request - objectChunks []Chunk - objectData []byte - ) - for { - req, err := stream.Recv() - if err == io.EOF { - break - } - if err != nil { - _ = stream.SendMsg(status.Errorf(codes.Internal, "%s: failed to recv message from stream: %v", op, err)) - return - } - switch data := req.Data.(type) { - case *plgpb.PutObjectRequest_Request: - if data.Request == nil { - _ = stream.SendMsg(status.Errorf(codes.InvalidArgument, "%s: missing request metadata", op)) - return - } - if data.Request.GetBucket() == nil { - _ = stream.SendMsg(status.Errorf(codes.InvalidArgument, "%s: missing storage bucket", op)) - return - } - if data.Request.GetBucket().GetSecrets() == nil { - _ = stream.SendMsg(status.Errorf(codes.InvalidArgument, "%s: missing secrets", op)) - return - } - if data.Request.Key == "" { - _ = stream.SendMsg(status.Errorf(codes.InvalidArgument, "%s; missing object key", op)) - return - } - request = data - case *plgpb.PutObjectRequest_FileChunk: - objectChunks = append(objectChunks, data.FileChunk) - objectData = append(objectData, data.FileChunk...) - case nil: - continue - default: - _ = stream.SendMsg(status.Errorf(codes.Internal, "%s: unknown message type: %v", op, data)) - return - } + if req.GetBucket() == nil { + return nil, status.Errorf(codes.InvalidArgument, "%s: missing storage bucket", op) + } + if req.GetBucket().GetBucketName() == "" { + return nil, status.Errorf(codes.InvalidArgument, "%s: missing bucket name", op) + } + if req.GetBucket().GetSecrets() == nil { + return nil, status.Errorf(codes.InvalidArgument, "%s: missing secrets", op) + } + if req.GetKey() == "" { + return nil, status.Errorf(codes.InvalidArgument, "%s; missing object key", op) + } + if req.GetPath() == "" { + return nil, status.Errorf(codes.InvalidArgument, "%s; missing path", op) + } + info, err := os.Stat(req.Path) + if err != nil { + return nil, status.Errorf(codes.InvalidArgument, "%s: file check failed: %v", op, err) + } + if info == nil { + return nil, status.Errorf(codes.Internal, "%s: failed to get file info", op) + } + if info.IsDir() { + return nil, status.Errorf(codes.InvalidArgument, "%s: path is a directory", op) + } + l.m.Lock() + defer l.m.Unlock() + bucket, ok := l.buckets[BucketName(req.GetBucket().GetBucketName())] + if !ok { + return nil, status.Errorf(codes.InvalidArgument, "%s: bucket not found", op) + } + for _, err := range l.errs { + if err.match(req.GetBucket(), req.GetKey()) { + return nil, status.Errorf(err.errCode, "%s: %s", op, err.errMsg) } - if request == nil { - _ = stream.SendMsg(status.Errorf(codes.InvalidArgument, "%s: request is nil", op)) - return - } - if len(objectData) <= 0 { - _ = stream.SendMsg(status.Errorf(codes.InvalidArgument, "%s: missing object data", op)) - return - } - l.m.Lock() - defer l.m.Unlock() - bucket, ok := l.buckets[BucketName(request.Request.GetBucket().GetBucketName())] - if !ok { - _ = stream.SendMsg(status.Errorf(codes.NotFound, "%s: bucket not found", op)) - return - } - for _, err := range l.errs { - if err.match(request.Request.GetBucket(), request.Request.GetKey()) { - _ = stream.SendMsg(status.Errorf(err.errCode, "%s: %s", op, err.errMsg)) - return - } - } - hash := sha256.New() - if _, err := io.Copy(hash, bytes.NewReader(objectData)); err != nil { - _ = stream.SendMsg(status.Errorf(codes.Internal, "%s: failed to hash object: %v", op, err)) - return - } - contentLength := int64(len(objectData)) - lastModified := time.Now() + } - // Ensure all the directories within the object key exist - parts := strings.Split(request.Request.GetKey(), "/") - tempPath := request.Request.GetBucket().GetBucketPrefix() - for _, p := range parts[:len(parts)-1] { - // Directories should have trailing `/` in the key - tempPath = fmt.Sprintf("%v%v/", tempPath, p) - bucket[ObjectName(tempPath)] = &storagePluginStorageInfo{ - lastModified: &lastModified, - } - } + lastModified := time.Now() + objectData, err := os.ReadFile(req.Path) + if err != nil { + return nil, status.Errorf(codes.Internal, "%s: failed to read file", op) + } + if len(objectData) == 0 { + return nil, status.Errorf(codes.InvalidArgument, "%s: missing object data", op) + } + contentLength := int64(len(objectData)) - // Now insert the object - objectPath := ObjectName(request.Request.GetBucket().GetBucketPrefix() + request.Request.GetKey()) - bucket[objectPath] = &storagePluginStorageInfo{ - DataChunks: objectChunks, - contentLength: &contentLength, - lastModified: &lastModified, + parts := strings.Split(req.GetKey(), "/") + tempPath := req.GetBucket().GetBucketPrefix() + for _, p := range parts[:len(parts)-1] { + // Directories should have trailing `/` in the key + tempPath = fmt.Sprintf("%v%v/", tempPath, p) + bucket[ObjectName(tempPath)] = &storagePluginStorageInfo{ + lastModified: &lastModified, } - if err := stream.SendAndClose(&plgpb.PutObjectResponse{ - ChecksumSha_256: hash.Sum(nil), - }); err != nil { - _ = stream.SendMsg(status.Errorf(codes.Internal, "%s: failed to close stream: %v", op, err)) - return + } + + objectChunks := []Chunk{} + for i := 0; i < len(objectData); i = i + l.chunksSize { + j := i + l.chunksSize + if j > len(objectData) { + j = len(objectData) } - }() - return nil + objectChunks = append(objectChunks, copyBytes(objectData[i:j])) + } + + // Now insert the object + objectPath := ObjectName(req.GetBucket().GetBucketPrefix() + req.GetKey()) + bucket[objectPath] = &storagePluginStorageInfo{ + DataChunks: objectChunks, + contentLength: &contentLength, + lastModified: &lastModified, + } + + hash := sha256.New() + if _, err := io.Copy(hash, bytes.NewReader(objectData)); err != nil { + return nil, status.Errorf(codes.Internal, "%s: failed to hash object: %v", op, err) + } + + return &plgpb.PutObjectResponse{ + ChecksumSha_256: hash.Sum(nil), + }, nil } func MockObject(data []Chunk) *storagePluginStorageInfo { diff --git a/internal/plugin/loopback/storage_test.go b/internal/plugin/loopback/storage_test.go index c7b66902a..533e8696c 100644 --- a/internal/plugin/loopback/storage_test.go +++ b/internal/plugin/loopback/storage_test.go @@ -8,6 +8,10 @@ import ( "context" "crypto/sha256" "io" + "io/fs" + "os" + "path" + "strings" "sync" "testing" @@ -666,62 +670,68 @@ func TestLoopbackGetObject(t *testing.T) { } func TestLoopbackPutObject(t *testing.T) { - require, assert := tr.New(t), ta.New(t) + require := tr.New(t) + td := t.TempDir() mockStorageMapData := map[BucketName]Bucket{ - "aws_s3_mock": {}, - "aws_s3_err": {}, + "object_store": {}, + "object_store_err": {}, } plg, err := NewLoopbackPlugin( WithMockBuckets(mockStorageMapData), WithMockError(PluginMockError{ - bucketName: "aws_s3_err", + bucketName: "object_store_err", objectKey: "mock_object", errMsg: "invalid credentials", errCode: codes.PermissionDenied, }), ) - assert.NoError(err) + require.NoError(err) + require.NotNil(plg) client := NewWrappingPluginStorageClient(plg) - secrets := &structpb.Struct{ - Fields: map[string]*structpb.Value{ - "AWS_ACCESS_KEY_ID": structpb.NewStringValue("access_key_id"), - "AWS_SECRET_ACCESS_KEY": structpb.NewStringValue("secret_access_key"), - }, - } - tests := []struct { - name string - request *plgpb.PutObjectRequest - dataChunks []Chunk - expectedErr codes.Code + name string + request *plgpb.PutObjectRequest + file *os.File + expectedErr codes.Code + expectedContent string }{ { - name: "missing request metadata", + name: "missing-request", + expectedErr: codes.InvalidArgument, + }, + { + name: "missing-bucket", + request: &plgpb.PutObjectRequest{}, + expectedErr: codes.InvalidArgument, + }, + { + name: "missing-bucket-name", request: &plgpb.PutObjectRequest{ - Data: &plgpb.PutObjectRequest_Request{}, + Bucket: &storagebuckets.StorageBucket{}, }, expectedErr: codes.InvalidArgument, }, { - name: "missing storage bucket", + name: "missing-secrets", request: &plgpb.PutObjectRequest{ - Data: &plgpb.PutObjectRequest_Request{ - Request: &plgpb.PutObjectMetadata{}, + Bucket: &storagebuckets.StorageBucket{ + BucketName: "object_store", }, }, expectedErr: codes.InvalidArgument, }, { - name: "missing secrets", + name: "missing-key", request: &plgpb.PutObjectRequest{ - Data: &plgpb.PutObjectRequest_Request{ - Request: &plgpb.PutObjectMetadata{ - Bucket: &storagebuckets.StorageBucket{ - BucketName: "aws_s3_mock", + Bucket: &storagebuckets.StorageBucket{ + BucketName: "object_store", + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + "SECRET_KEY_ID": structpb.NewStringValue("secret_key_id"), }, }, }, @@ -729,210 +739,210 @@ func TestLoopbackPutObject(t *testing.T) { expectedErr: codes.InvalidArgument, }, { - name: "missing object key", + name: "missing-path", request: &plgpb.PutObjectRequest{ - Data: &plgpb.PutObjectRequest_Request{ - Request: &plgpb.PutObjectMetadata{ - Bucket: &storagebuckets.StorageBucket{ - BucketName: "aws_s3_mock", - Secrets: secrets, + Bucket: &storagebuckets.StorageBucket{ + BucketName: "object_store", + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + "SECRET_KEY_ID": structpb.NewStringValue("secret_key_id"), }, }, }, + Key: "test-file", }, expectedErr: codes.InvalidArgument, }, { - name: "empty object data", - dataChunks: []Chunk{ - []byte(""), - }, + name: "file-not-found", request: &plgpb.PutObjectRequest{ - Data: &plgpb.PutObjectRequest_Request{ - Request: &plgpb.PutObjectMetadata{ - Bucket: &storagebuckets.StorageBucket{ - BucketName: "aws_s3_mock", - Secrets: secrets, + Bucket: &storagebuckets.StorageBucket{ + BucketName: "object_store", + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + "SECRET_KEY_ID": structpb.NewStringValue("secret_key_id"), }, - Key: "mock_object", }, }, + Key: "test-file", + Path: path.Join(td, "test-file"), }, expectedErr: codes.InvalidArgument, }, { - name: "bucket not found", - dataChunks: []Chunk{ - []byte("THIS IS A MOCKED OBJECT"), - }, + name: "path-is-directory", request: &plgpb.PutObjectRequest{ - Data: &plgpb.PutObjectRequest_Request{ - Request: &plgpb.PutObjectMetadata{ - Bucket: &storagebuckets.StorageBucket{ - BucketName: "invalid_bucket", - Secrets: secrets, + Bucket: &storagebuckets.StorageBucket{ + BucketName: "object_store", + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + "SECRET_KEY_ID": structpb.NewStringValue("secret_key_id"), }, - Key: "mock_object", }, }, + Key: "test-directory", + Path: path.Join(td, "test-directory"), }, - expectedErr: codes.NotFound, + file: func() *os.File { + err := os.Mkdir(path.Join(td, "test-directory"), fs.ModeAppend) + require.NoError(err) + return nil + }(), + expectedErr: codes.InvalidArgument, }, { - name: "mock error", - dataChunks: []Chunk{ - []byte("THIS IS A MOCKED OBJECT"), - }, + name: "bucket-not-found", request: &plgpb.PutObjectRequest{ - Data: &plgpb.PutObjectRequest_Request{ - Request: &plgpb.PutObjectMetadata{ - Bucket: &storagebuckets.StorageBucket{ - BucketName: "aws_s3_err", - Secrets: secrets, + Bucket: &storagebuckets.StorageBucket{ + BucketName: "object_store_dne", + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + "SECRET_KEY_ID": structpb.NewStringValue("secret_key_id"), }, - Key: "mock_object", }, }, + Key: "test-bucket-not-found", + Path: path.Join(td, "test-bucket-not-found"), }, + file: func() *os.File { + file, err := os.Create(path.Join(td, "test-bucket-not-found")) + require.NoError(err) + return file + }(), + expectedErr: codes.InvalidArgument, + }, + { + name: "mock-error", + request: &plgpb.PutObjectRequest{ + Bucket: &storagebuckets.StorageBucket{ + BucketName: "object_store_err", + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + "SECRET_KEY_ID": structpb.NewStringValue("secret_key_id"), + }, + }, + }, + Key: "mock_object", + Path: path.Join(td, "test-bucket-not-found"), + }, + file: func() *os.File { + file, err := os.Create(path.Join(td, "test-mock-error")) + require.NoError(err) + return file + }(), expectedErr: codes.PermissionDenied, }, { - name: "valid object", - dataChunks: []Chunk{ - []byte("THIS IS A MOCKED OBJECT"), - }, + name: "emtpy-object-data", request: &plgpb.PutObjectRequest{ - Data: &plgpb.PutObjectRequest_Request{ - Request: &plgpb.PutObjectMetadata{ - Bucket: &storagebuckets.StorageBucket{ - BucketName: "aws_s3_mock", - Secrets: secrets, + Bucket: &storagebuckets.StorageBucket{ + BucketName: "object_store", + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + "SECRET_KEY_ID": structpb.NewStringValue("secret_key_id"), }, - Key: "mock_object", }, }, + Key: "test-empty-object-data", + Path: path.Join(td, "test-empty-object-data"), }, + file: func() *os.File { + file, err := os.Create(path.Join(td, "test-empty-object-data")) + require.NoError(err) + require.NoError(file.Close()) + return file + }(), + expectedErr: codes.InvalidArgument, }, { name: "valid object with dir in key", - dataChunks: []Chunk{ - []byte("THIS IS A MOCKED OBJECT"), - }, request: &plgpb.PutObjectRequest{ - Data: &plgpb.PutObjectRequest_Request{ - Request: &plgpb.PutObjectMetadata{ - Bucket: &storagebuckets.StorageBucket{ - BucketName: "aws_s3_mock", - Secrets: secrets, + Bucket: &storagebuckets.StorageBucket{ + BucketName: "object_store", + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + "SECRET_KEY_ID": structpb.NewStringValue("secret_key_id"), }, - Key: "foo/bar/zoo/mocked_object", }, }, + Key: "foo/bar/zoo/mocked_object", + Path: path.Join(td, "test-valid-object-dir-in-key"), }, + file: func() *os.File { + file, err := os.Create(path.Join(td, "test-valid-object-dir-in-key")) + require.NoError(err) + n, err := file.WriteString("TEST OBJ WITH DIR IN KEY!") + require.NoError(err) + require.Equal(len("TEST OBJ WITH DIR IN KEY!"), n) + require.NoError(file.Close()) + return file + }(), + expectedContent: "TEST OBJ WITH DIR IN KEY!", }, { name: "valid object w/ prefix", - dataChunks: []Chunk{ - []byte("THIS IS A MOCKED OBJECT"), - }, request: &plgpb.PutObjectRequest{ - Data: &plgpb.PutObjectRequest_Request{ - Request: &plgpb.PutObjectMetadata{ - Bucket: &storagebuckets.StorageBucket{ - BucketName: "aws_s3_mock", - BucketPrefix: "/filtered/", - Secrets: secrets, + Bucket: &storagebuckets.StorageBucket{ + BucketName: "object_store", + BucketPrefix: "/filtered/", + Secrets: &structpb.Struct{ + Fields: map[string]*structpb.Value{ + "SECRET_KEY_ID": structpb.NewStringValue("secret_key_id"), }, - Key: "mock_object", - }, - }, - }, - }, - { - name: "valid object w/ multiple chunks", - dataChunks: []Chunk{ - []byte("PART A: 1234567890"), - []byte("PART B: 0987654321"), - []byte("PART C: qwertyuiop"), - []byte("PART D: poiuytrewq"), - []byte("PART E: asdfghjkl"), - []byte("PART F: lkjhgfdsa"), - []byte("PART G: zxcvbnm"), - []byte("PART H: mnbvcxz"), - }, - request: &plgpb.PutObjectRequest{ - Data: &plgpb.PutObjectRequest_Request{ - Request: &plgpb.PutObjectMetadata{ - Bucket: &storagebuckets.StorageBucket{ - BucketName: "aws_s3_mock", - Secrets: secrets, - }, - Key: "multi_chunk_object", }, }, + Key: "mocked_object", + Path: path.Join(td, "test-valid-object-prefix"), }, + file: func() *os.File { + file, err := os.Create(path.Join(td, "test-valid-object-prefix")) + require.NoError(err) + n, err := file.WriteString("TEST OBJ WITH PREFIX!") + require.NoError(err) + require.Equal(len("TEST OBJ WITH PREFIX!"), n) + require.NoError(file.Close()) + return file + }(), + expectedContent: "TEST OBJ WITH PREFIX!", }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - stream, err := client.PutObject(context.Background()) - assert.NoError(err) - assert.NotNil(stream) - - var objectData []byte - var closeResponse *plgpb.PutObjectResponse - var closeErr error - var wg sync.WaitGroup - wg.Add(1) - go func() { - defer wg.Done() - err = stream.Send(tt.request) - assert.NoError(err) - - for _, chunk := range tt.dataChunks { - objectData = append(objectData, chunk...) - err = stream.Send(&plgpb.PutObjectRequest{ - Data: &plgpb.PutObjectRequest_FileChunk{ - FileChunk: chunk, - }, - }) - assert.NoError(err) - } - - closeResponse, closeErr = stream.CloseAndRecv() - }() - wg.Wait() - + resp, err := client.PutObject(context.Background(), tt.request) if tt.expectedErr != codes.OK { - assert.Error(closeErr) - assert.Nil(closeResponse) - s, ok := status.FromError(closeErr) + require.Error(err) + require.Nil(resp) + s, ok := status.FromError(err) require.True(ok, "invalid error type") - assert.Equal(s.Code(), tt.expectedErr) + require.Equal(s.Code(), tt.expectedErr) return } + require.NoError(err) + require.NotNil(resp) - assert.NoError(closeErr) - require.NotNil(closeResponse) - - var actualObject *storagePluginStorageInfo - objectPath := ObjectName(tt.request.GetRequest().GetBucket().GetBucketPrefix() + tt.request.GetRequest().GetKey()) - if obj, ok := mockStorageMapData[BucketName(tt.request.GetRequest().GetBucket().GetBucketName())][objectPath]; ok { - actualObject = obj + actualBucket, ok := plg.buckets[BucketName(tt.request.Bucket.BucketName)] + require.True(ok) + require.NotEmpty(actualBucket) + actualObject, ok := actualBucket[ObjectName(path.Join(tt.request.Bucket.BucketPrefix, tt.request.Key))] + require.True(ok) + require.NotEmpty(actualObject) + actualData := []byte{} + for _, c := range actualObject.DataChunks { + actualData = append(actualData, c...) } - require.NotNil(actualObject) - assert.EqualValues(tt.dataChunks, actualObject.DataChunks) - assert.NotEmpty(closeResponse.ChecksumSha_256) + require.Equal(tt.expectedContent, string(actualData)) hash := sha256.New() - _, err = io.Copy(hash, bytes.NewReader(objectData)) + _, err = io.Copy(hash, bytes.NewReader(actualData)) require.NoError(err) - assert.Equal(hash.Sum(nil), closeResponse.ChecksumSha_256) + require.ElementsMatch(hash.Sum(nil), resp.ChecksumSha_256) }) } } func TestLoopbackStoragePlugin(t *testing.T) { + td := t.TempDir() + require, assert := tr.New(t), ta.New(t) mockStorageMapData := map[BucketName]Bucket{ @@ -951,38 +961,31 @@ func TestLoopbackStoragePlugin(t *testing.T) { }, } + objectData := "THIS IS A MOCKED OBJECT" bucket := &storagebuckets.StorageBucket{ BucketName: "aws_s3_mock", BucketPrefix: "/filtered/path/", Secrets: secrets, } - objectData := []byte("THIS IS A MOCKED OBJECT") - putObjectStream, err := client.PutObject(context.Background()) - assert.NoError(err) - - err = putObjectStream.Send(&plgpb.PutObjectRequest{ - Data: &plgpb.PutObjectRequest_Request{ - Request: &plgpb.PutObjectMetadata{ - Bucket: bucket, - Key: "dir1/mock_object", - }, - }, - }) - assert.NoError(err) - err = putObjectStream.Send(&plgpb.PutObjectRequest{ - Data: &plgpb.PutObjectRequest_FileChunk{ - FileChunk: objectData, - }, - }) - assert.NoError(err) - - putResponse, err := putObjectStream.CloseAndRecv() + file, err := os.Create(path.Join(td, "test-put-object")) require.NoError(err) + n, err := file.WriteString(objectData) + require.NoError(err) + require.Equal(len(objectData), n) + require.NoError(file.Close()) + + putResponse, err := client.PutObject( + context.Background(), &plgpb.PutObjectRequest{ + Bucket: bucket, + Key: "dir1/mock_object", + Path: path.Join(td, "test-put-object"), + }) + assert.NoError(err) require.NotNil(putResponse) assert.NotEmpty(putResponse.GetChecksumSha_256()) hash := sha256.New() - _, err = io.Copy(hash, bytes.NewReader(objectData)) + _, err = io.Copy(hash, strings.NewReader(objectData)) require.NoError(err) assert.EqualValues(hash.Sum(nil), putResponse.GetChecksumSha_256()) diff --git a/internal/plugin/loopback/testing_grpc_stream.go b/internal/plugin/loopback/testing_grpc_stream.go index 23e0b0e96..21d37b539 100644 --- a/internal/plugin/loopback/testing_grpc_stream.go +++ b/internal/plugin/loopback/testing_grpc_stream.go @@ -229,274 +229,3 @@ func newGetObjectStream() *getObjectStream { } return stream } - -// putObjectStreamRequest is used to mock a message sent from the client to the server. -type putObjectStreamRequest struct { - msg *plgpb.PutObjectRequest - err error -} - -// puObjectStreamResponse is used to mock a message sent from the server to the client. -type putObjectStreamResponse struct { - msg *plgpb.PutObjectResponse - err error -} - -// putObjectStream is used to mock the interactions between -// the client and server for the PutObject method. -type putObjectStream struct { - client plgpb.StoragePluginService_PutObjectClient - server plgpb.StoragePluginService_PutObjectServer - - // requests is used to mock the client sending messages to the server. - requests chan *putObjectStreamRequest - - // responses is used to mock the server sending messages to the client. - responses chan *putObjectStreamResponse - - m *sync.Mutex - clientClosed bool - serverClosed bool -} - -func (s *putObjectStream) CloseClient() { - s.m.Lock() - defer s.m.Unlock() - if s.clientClosed { - return - } - close(s.requests) - s.clientClosed = true -} - -func (s *putObjectStream) IsClientClosed() bool { - s.m.Lock() - defer s.m.Unlock() - return s.clientClosed -} - -func (s *putObjectStream) CloseServer() { - s.m.Lock() - defer s.m.Unlock() - if s.serverClosed { - return - } - close(s.responses) - s.serverClosed = true -} - -func (s *putObjectStream) IsServerClosed() bool { - s.m.Lock() - defer s.m.Unlock() - return s.serverClosed -} - -// putObjectClient is used to mock the client stream -// interactions for the PutObject method. -type putObjectClient struct { - sendToServer chan *putObjectStreamRequest - sentFromServer chan *putObjectStreamResponse - - // closeStream is used to close the channels of the stream. - // This is used to prevent the channels from being closed multiple times. - // This is needed because the channel can be closed by the client or the server. - closeStream func() - - // isStreamClosed is used to check if the stream is closed. - // This is needed because the channel can be closed by the client or the server. - isStreamClosed func() bool -} - -// Send will send a message to the server. -// Send will return an error if the request is nil. -// Send will return an error if the stream is closed. -func (c *putObjectClient) Send(req *plgpb.PutObjectRequest) error { - if req == nil { - return fmt.Errorf(`parameter arg "req PutObjectRequest" cannot be nil`) - } - if c.isStreamClosed() { - return fmt.Errorf("stream is closed") - } - c.sendToServer <- &putObjectStreamRequest{msg: req} - return nil -} - -// CloseAndRecv will return an PutObjectResponse if the server returns one. -// CloseAndRecv will return an error if the server returns an error. -// CloseAndRecv will return an io.EOF error if the server closes the stream. -// CloseAndRecv will close the stream used to send messages to the server. -func (c *putObjectClient) CloseAndRecv() (*plgpb.PutObjectResponse, error) { - c.closeStream() - resp, ok := <-c.sentFromServer - if !ok { - return nil, io.EOF - } - return resp.msg, resp.err -} - -// Header should not be used. -// Header is implemeted to satisfy the grpc.ClientStream interface. -// Header will always return an empty metadata and an nil error. -func (c *putObjectClient) Header() (metadata.MD, error) { - return make(metadata.MD), nil -} - -// Trailer should not be used. -// Trailer is implemeted to satisfy the grpc.ClientStream interface. -// Trailer will always return an empty metadata. -func (c *putObjectClient) Trailer() metadata.MD { - return make(metadata.MD) -} - -// CloseSend will close the channel used to retrieve messages from -// the server. This will cause the CloseAndRecv method to return io.EOF. -// CloseSend will always return a nill error. -func (c *putObjectClient) CloseSend() error { - if c.isStreamClosed() { - return fmt.Errorf("stream is closed") - } - c.closeStream() - return nil -} - -// Context will always return a Background context. -func (c *putObjectClient) Context() context.Context { - return context.Background() -} - -// SendMsg should not be used. -// SendMsg is implemeted to satisfy the grpc.ClientStream interface. -// SendMsg will always return an nil error. -func (c *putObjectClient) SendMsg(m interface{}) error { - return nil -} - -// RecvMsg should not be used. -// RecvMsg is implemeted to satisfy the grpc.ClientStream interface. -// RecvMsg will always return an nil error. -func (c *putObjectClient) RecvMsg(m interface{}) error { - return nil -} - -// putObjectServer is used to mock the server stream -// interactions for the PutObject method. -type putObjectServer struct { - sentFromClient chan *putObjectStreamRequest - sentToClient chan *putObjectStreamResponse - - // closeStream is used to close the channels of the stream. - // This is used to prevent the channels from being closed multiple times. - // This is needed because the channel can be closed by the client or the server. - closeStream func() - - // isStreamClosed is used to check if the stream is closed. - // This is needed because the channel can be closed by the client or the server. - isStreamClosed func() bool -} - -// SendAndClose will send a PutObjectResponse to the client. -// SendAndClose will return an error if the response is nil. -// SendAndClose will close the stream used to send messages from the client. -// SendAndClose will return an error if the stream is closed. -func (s *putObjectServer) SendAndClose(resp *plgpb.PutObjectResponse) error { - if resp == nil { - return fmt.Errorf(`parameter arg "resp PutObjectResponse" cannot be nil`) - } - if s.isStreamClosed() { - return fmt.Errorf("stream is closed") - } - s.sentToClient <- &putObjectStreamResponse{msg: resp} - s.closeStream() - return nil -} - -// Recv will read a PutObjectRequest from the stream. -// Recv will return an io.EOF error if the stream is closed. -func (s *putObjectServer) Recv() (*plgpb.PutObjectRequest, error) { - req, ok := <-s.sentFromClient - if !ok { - return nil, io.EOF - } - return req.msg, req.err -} - -// SetHeader should not be used. -// SetHeader is implemeted to satisfy the grpc.ServerStream interface. -// SetHeader will always return an nil error. -func (s *putObjectServer) SetHeader(metadata.MD) error { - return nil -} - -// SendHeader should not be used. -// SendHeader is implemeted to satisfy the grpc.ServerStream interface. -// SendHeader will always return an nil error. -func (s *putObjectServer) SendHeader(metadata.MD) error { - return nil -} - -// SetTrailer should not be used. -// SetTrailer is implemeted to satisfy the grpc.ServerStream interface. -func (s *putObjectServer) SetTrailer(metadata.MD) { -} - -// Context will always return a Background context. -func (s *putObjectServer) Context() context.Context { - return context.Background() -} - -// SendMsg allows sending errors other than io.EOF to the client. -// Sending an error message will close the stream. -// SendMsg allows sending PutObjectResponse messages to the client. -// SendMsg returns an invalid argument error if the message is not -// an error or PutObjectResponse. -// SendMsg will return an error if the stream is closed. -func (s *putObjectServer) SendMsg(m interface{}) error { - switch msg := m.(type) { - case *plgpb.PutObjectResponse: - if s.isStreamClosed() { - return fmt.Errorf("stream is closed") - } - s.sentToClient <- &putObjectStreamResponse{msg: msg} - case error: - if s.isStreamClosed() { - return fmt.Errorf("stream is closed") - } - s.sentToClient <- &putObjectStreamResponse{err: msg} - defer s.closeStream() - default: - return fmt.Errorf("invalid argument %v", m) - } - return nil -} - -// RecvMsg should not be used. -// RecvMsg is implemeted to satisfy the grpc.ServerStream interface. -// RecvMsg will always return an nil error. -func (s *putObjectServer) RecvMsg(m interface{}) error { - return nil -} - -// newPutObjectStream will create a mock stream for the PutObject method. -// The client and server stream is mocked by creating a PutObjectRequest, -// PutObjectResponse and an error channel that is shared between the client -// and server. -func newPutObjectStream() *putObjectStream { - stream := &putObjectStream{ - m: new(sync.Mutex), - requests: make(chan *putObjectStreamRequest), - responses: make(chan *putObjectStreamResponse), - } - stream.client = &putObjectClient{ - sendToServer: stream.requests, - sentFromServer: stream.responses, - closeStream: stream.CloseClient, - isStreamClosed: stream.IsClientClosed, - } - stream.server = &putObjectServer{ - sentFromClient: stream.requests, - sentToClient: stream.responses, - closeStream: stream.CloseServer, - isStreamClosed: stream.IsServerClosed, - } - return stream -} diff --git a/internal/plugin/loopback/testing_grpc_stream_test.go b/internal/plugin/loopback/testing_grpc_stream_test.go index dd5aec9e7..8425fcda8 100644 --- a/internal/plugin/loopback/testing_grpc_stream_test.go +++ b/internal/plugin/loopback/testing_grpc_stream_test.go @@ -227,242 +227,6 @@ func TestNewGetObjectStream(t *testing.T) { require.False(ok) } -func Test_PutObjectStream_Client(t *testing.T) { - require := tr.New(t) - - stream := newPutObjectStream() - require.NotNil(stream) - require.NotNil(stream.client) - - // Validate Header is not nil and does not return an error - header, err := stream.client.Header() - require.NotNil(header) - require.NoError(err) - - // Validate Trailer is not nil - trailer := stream.client.Trailer() - require.NotNil(trailer) - - // Validate Context is not nil - ctx := stream.client.Context() - require.NotNil(ctx) - - // Validate SendMsg does not write to the channel requests - err = stream.client.SendMsg(nil) - require.NoError(err) - require.Empty(stream.requests) - - // Validate RecvMsg does not read from the channel requests - streamSize := len(stream.requests) - var msg *putObjectStreamResponse - err = stream.client.RecvMsg(msg) - require.NoError(err) - require.Nil(msg) - require.Equal(streamSize, len(stream.requests)) - - // Validate Send returns an error when the request is nil - err = stream.client.Send(nil) - require.Error(err) - - // Validate Send writes to the channel requests - var wg sync.WaitGroup - go func() { - err = stream.client.Send(&plgpb.PutObjectRequest{}) - require.NoError(err) - }() - wg.Add(1) - go func() { - defer wg.Done() - req, ok := <-stream.requests - require.True(ok) - require.NotNil(req) - require.NotNil(req.msg) - require.Nil(req.err) - }() - wg.Wait() - - // Validate CloseSend does not return an error - // Validate clientClosed is set to true - // Validate the channel requests is closed - var ok bool - err = stream.client.CloseSend() - require.NoError(err) - require.True(stream.clientClosed) - _, ok = <-stream.requests - require.False(ok) - - // Validate CloseSend does return an error when the channel is closed - // Validate clientClosed is set to true - // Validate the channel requests is closed - err = stream.client.CloseSend() - require.Error(err) - require.Contains(err.Error(), "stream is closed") - require.True(stream.clientClosed) - _, ok = <-stream.requests - require.False(ok) - - // Validate Send returns an error when the channel is closed - err = stream.client.Send(&plgpb.PutObjectRequest{}) - require.Error(err) - require.Contains(err.Error(), "stream is closed") - require.Empty(stream.requests) - - // Reset stream - stream = newPutObjectStream() - - // Validate CloseAndRecv returns an EOF error when the channel responses is closed - // Validate CloseAndRecv closes the channel requests - stream.CloseServer() - resp, err := stream.client.CloseAndRecv() - require.Equal(io.EOF, err) - require.Nil(resp) - require.True(stream.clientClosed) - require.True(stream.IsClientClosed()) - require.True(stream.serverClosed) - require.True(stream.IsServerClosed()) -} - -func Test_PutObjectStream_Server(t *testing.T) { - require := tr.New(t) - - stream := newPutObjectStream() - require.NotNil(stream) - require.NotNil(stream.server) - - // Validate SetHeader does not return an error - err := stream.server.SetHeader(nil) - require.NoError(err) - - // Validate SendHeader does not return an error - err = stream.server.SendHeader(nil) - require.NoError(err) - - // Validate Context returns a context - ctx := stream.server.Context() - require.NotNil(ctx) - - // Validate RecvMsg does not read from the channel requests - streamSize := len(stream.requests) - var msg *putObjectStreamRequest - err = stream.server.RecvMsg(msg) - require.NoError(err) - require.Nil(msg) - require.Equal(streamSize, len(stream.requests)) - - // Validate SendMsg returns an error for unknown message types - err = stream.server.SendMsg(map[string]string{}) - require.Error(err) - require.Contains(err.Error(), "invalid argument") - - // Validate SendMsg does write a message to the channel responses - var wg sync.WaitGroup - go func() { - err = stream.server.SendMsg(&plgpb.PutObjectResponse{}) - require.NoError(err) - }() - wg.Add(1) - go func() { - defer wg.Done() - resp, ok := <-stream.responses - require.True(ok) - require.NotNil(resp) - require.NotNil(resp.msg) - require.Nil(resp.err) - }() - wg.Wait() - - // Validate SendMsg does write a error to the channel responses - // Validate SendMsg closes the channel responses - go func() { - err = stream.server.SendMsg(fmt.Errorf("mock error")) - require.NoError(err) - }() - wg.Add(1) - go func() { - defer wg.Done() - resp, ok := <-stream.responses - require.True(ok) - require.NotNil(resp) - require.NotNil(resp.err) - require.Nil(resp.msg) - }() - wg.Wait() - _, ok := <-stream.responses - require.False(ok) - require.True(stream.serverClosed) - require.True(stream.IsServerClosed()) - - // Validate SendMsg does return an error when the channel responses is closed - err = stream.server.SendMsg(&plgpb.PutObjectResponse{}) - require.Error(err) - require.Contains(err.Error(), "stream is closed") - err = stream.server.SendMsg(fmt.Errorf("mock error")) - require.Error(err) - require.Contains(err.Error(), "stream is closed") - - // Validate SendAndClose returns an error when response is nil - err = stream.server.SendAndClose(nil) - require.Error(err) - - // Validate SendAndClose returns an error when the channel responses is closed - err = stream.server.SendAndClose(&plgpb.PutObjectResponse{}) - require.Error(err) - require.Contains(err.Error(), "stream is closed") - - // Reset stream - stream = newPutObjectStream() - - // Validate SendAndClose closes the responses channel - go func() { - err = stream.server.SendAndClose(&plgpb.PutObjectResponse{}) - require.NoError(err) - }() - wg.Add(1) - go func() { - defer wg.Done() - resp, ok := <-stream.responses - require.True(ok) - require.NotNil(resp) - require.NotNil(resp.msg) - require.Nil(resp.err) - }() - wg.Wait() - require.True(stream.serverClosed) - require.True(stream.IsServerClosed()) - _, ok = <-stream.responses - require.False(ok) -} - -func TestNewPutObjectStream(t *testing.T) { - require := tr.New(t) - - stream := newPutObjectStream() - require.NotNil(stream) - require.NotNil(stream.m) - require.NotNil(stream.client) - require.NotNil(stream.server) - require.NotNil(stream.requests) - require.NotNil(stream.responses) - require.False(stream.clientClosed) - require.False(stream.IsClientClosed()) - require.False(stream.serverClosed) - require.False(stream.IsServerClosed()) - - // Validate CloseClient() closes the requests channel - stream.CloseClient() - require.True(stream.clientClosed) - require.True(stream.IsClientClosed()) - _, ok := <-stream.requests - require.False(ok) - - // Validate CloseServer() closes the responses channel - stream.CloseServer() - require.True(stream.serverClosed) - require.True(stream.IsServerClosed()) - _, ok = <-stream.responses - require.False(ok) -} - func Test_GetObjectStream(t *testing.T) { require := tr.New(t) @@ -508,95 +272,3 @@ func Test_GetObjectStream(t *testing.T) { _, ok = <-stream.messages require.False(ok) } - -func Test_PutObjectStream(t *testing.T) { - require := tr.New(t) - - stream := newPutObjectStream() - - // Validate the client recieves the response when the server closes the stream - var wg sync.WaitGroup - wg.Add(2) - go func() { - defer wg.Done() - for { - resp, err := stream.client.CloseAndRecv() - require.NotNil(resp) - require.NoError(err) - break - } - }() - go func() { - defer wg.Done() - err := stream.server.SendAndClose(&plgpb.PutObjectResponse{}) - require.NoError(err) - }() - wg.Wait() - require.True(stream.clientClosed) - require.True(stream.IsClientClosed()) - require.True(stream.serverClosed) - require.True(stream.IsServerClosed()) - _, ok := <-stream.requests - require.False(ok) - _, ok = <-stream.responses - require.False(ok) - - // Reset Stream - stream = newPutObjectStream() - - // Validate the client recieves an error when the server sends an error - wg.Add(2) - go func() { - defer wg.Done() - for { - resp, err := stream.client.CloseAndRecv() - require.Nil(resp) - require.Error(err) - require.Contains(err.Error(), "mock error") - break - } - }() - go func() { - defer wg.Done() - err := stream.server.SendMsg(fmt.Errorf("mock error")) - require.NoError(err) - }() - wg.Wait() - require.True(stream.clientClosed) - require.True(stream.IsClientClosed()) - require.True(stream.serverClosed) - require.True(stream.IsServerClosed()) - _, ok = <-stream.requests - require.False(ok) - _, ok = <-stream.responses - require.False(ok) - - // Reset Stream - stream = newPutObjectStream() - - // Validate the server recieves an EOF error when the client closes the stream - wg.Add(2) - go func() { - defer wg.Done() - for { - req, err := stream.server.Recv() - if err != nil { - require.Nil(req) - require.Equal(io.EOF, err) - break - } - } - }() - go func() { - wg.Done() - err := stream.client.CloseSend() - require.NoError(err) - }() - wg.Wait() - require.True(stream.clientClosed) - require.True(stream.IsClientClosed()) - require.False(stream.serverClosed) - require.False(stream.IsServerClosed()) - _, ok = <-stream.requests - require.False(ok) -} diff --git a/internal/proto/plugin/v1/storage_plugin_service.proto b/internal/proto/plugin/v1/storage_plugin_service.proto index 9a3432f87..3a5840faf 100644 --- a/internal/proto/plugin/v1/storage_plugin_service.proto +++ b/internal/proto/plugin/v1/storage_plugin_service.proto @@ -35,9 +35,9 @@ service StoragePluginService { // GetObject is a hook that retrieves objects. rpc GetObject(GetObjectRequest) returns (stream GetObjectResponse); - // PutObject is a hook that streams chunks of a file to be stored - // in an external object store. - rpc PutObject(stream PutObjectRequest) returns (PutObjectResponse); + // PutObject is a hook that reads a file stored on local disk and + // stores it to an external object store. + rpc PutObject(PutObjectRequest) returns (PutObjectResponse); } message OnCreateStorageBucketRequest { @@ -116,22 +116,15 @@ message GetObjectResponse { bytes file_chunk = 10; } -message PutObjectMetadata { +message PutObjectRequest { // Required. The existing state of the storage bucket. controller.api.resources.storagebuckets.v1.StorageBucket bucket = 10; // Required. The path of the object. string key = 20; -} -message PutObjectRequest { - oneof data { - // Object data. - // Recommended chunk size should be between 16KiB to 64KiB. - bytes file_chunk = 10; - - PutObjectMetadata request = 20; - } + // Required. The path of the object on local disk. + string path = 30; } message PutObjectResponse { diff --git a/sdk/pbs/plugin/storage_plugin_service.pb.go b/sdk/pbs/plugin/storage_plugin_service.pb.go index 2aa518aa3..8331336aa 100644 --- a/sdk/pbs/plugin/storage_plugin_service.pb.go +++ b/sdk/pbs/plugin/storage_plugin_service.pb.go @@ -640,7 +640,7 @@ func (x *GetObjectResponse) GetFileChunk() []byte { return nil } -type PutObjectMetadata struct { +type PutObjectRequest struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields @@ -649,70 +649,14 @@ type PutObjectMetadata struct { Bucket *storagebuckets.StorageBucket `protobuf:"bytes,10,opt,name=bucket,proto3" json:"bucket,omitempty"` // Required. The path of the object. Key string `protobuf:"bytes,20,opt,name=key,proto3" json:"key,omitempty"` -} - -func (x *PutObjectMetadata) Reset() { - *x = PutObjectMetadata{} - if protoimpl.UnsafeEnabled { - mi := &file_plugin_v1_storage_plugin_service_proto_msgTypes[12] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) - } -} - -func (x *PutObjectMetadata) String() string { - return protoimpl.X.MessageStringOf(x) -} - -func (*PutObjectMetadata) ProtoMessage() {} - -func (x *PutObjectMetadata) ProtoReflect() protoreflect.Message { - mi := &file_plugin_v1_storage_plugin_service_proto_msgTypes[12] - if protoimpl.UnsafeEnabled && x != nil { - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - if ms.LoadMessageInfo() == nil { - ms.StoreMessageInfo(mi) - } - return ms - } - return mi.MessageOf(x) -} - -// Deprecated: Use PutObjectMetadata.ProtoReflect.Descriptor instead. -func (*PutObjectMetadata) Descriptor() ([]byte, []int) { - return file_plugin_v1_storage_plugin_service_proto_rawDescGZIP(), []int{12} -} - -func (x *PutObjectMetadata) GetBucket() *storagebuckets.StorageBucket { - if x != nil { - return x.Bucket - } - return nil -} - -func (x *PutObjectMetadata) GetKey() string { - if x != nil { - return x.Key - } - return "" -} - -type PutObjectRequest struct { - state protoimpl.MessageState - sizeCache protoimpl.SizeCache - unknownFields protoimpl.UnknownFields - - // Types that are assignable to Data: - // - // *PutObjectRequest_FileChunk - // *PutObjectRequest_Request - Data isPutObjectRequest_Data `protobuf_oneof:"data"` + // Required. The path of the object on local disk. + Path string `protobuf:"bytes,30,opt,name=path,proto3" json:"path,omitempty"` } func (x *PutObjectRequest) Reset() { *x = PutObjectRequest{} if protoimpl.UnsafeEnabled { - mi := &file_plugin_v1_storage_plugin_service_proto_msgTypes[13] + mi := &file_plugin_v1_storage_plugin_service_proto_msgTypes[12] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -725,7 +669,7 @@ func (x *PutObjectRequest) String() string { func (*PutObjectRequest) ProtoMessage() {} func (x *PutObjectRequest) ProtoReflect() protoreflect.Message { - mi := &file_plugin_v1_storage_plugin_service_proto_msgTypes[13] + mi := &file_plugin_v1_storage_plugin_service_proto_msgTypes[12] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -738,48 +682,30 @@ func (x *PutObjectRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use PutObjectRequest.ProtoReflect.Descriptor instead. func (*PutObjectRequest) Descriptor() ([]byte, []int) { - return file_plugin_v1_storage_plugin_service_proto_rawDescGZIP(), []int{13} + return file_plugin_v1_storage_plugin_service_proto_rawDescGZIP(), []int{12} } -func (m *PutObjectRequest) GetData() isPutObjectRequest_Data { - if m != nil { - return m.Data +func (x *PutObjectRequest) GetBucket() *storagebuckets.StorageBucket { + if x != nil { + return x.Bucket } return nil } -func (x *PutObjectRequest) GetFileChunk() []byte { - if x, ok := x.GetData().(*PutObjectRequest_FileChunk); ok { - return x.FileChunk +func (x *PutObjectRequest) GetKey() string { + if x != nil { + return x.Key } - return nil + return "" } -func (x *PutObjectRequest) GetRequest() *PutObjectMetadata { - if x, ok := x.GetData().(*PutObjectRequest_Request); ok { - return x.Request +func (x *PutObjectRequest) GetPath() string { + if x != nil { + return x.Path } - return nil + return "" } -type isPutObjectRequest_Data interface { - isPutObjectRequest_Data() -} - -type PutObjectRequest_FileChunk struct { - // Object data. - // Recommended chunk size should be between 16KiB to 64KiB. - FileChunk []byte `protobuf:"bytes,10,opt,name=file_chunk,json=fileChunk,proto3,oneof"` -} - -type PutObjectRequest_Request struct { - Request *PutObjectMetadata `protobuf:"bytes,20,opt,name=request,proto3,oneof"` -} - -func (*PutObjectRequest_FileChunk) isPutObjectRequest_Data() {} - -func (*PutObjectRequest_Request) isPutObjectRequest_Data() {} - type PutObjectResponse struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache @@ -792,7 +718,7 @@ type PutObjectResponse struct { func (x *PutObjectResponse) Reset() { *x = PutObjectResponse{} if protoimpl.UnsafeEnabled { - mi := &file_plugin_v1_storage_plugin_service_proto_msgTypes[14] + mi := &file_plugin_v1_storage_plugin_service_proto_msgTypes[13] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -805,7 +731,7 @@ func (x *PutObjectResponse) String() string { func (*PutObjectResponse) ProtoMessage() {} func (x *PutObjectResponse) ProtoReflect() protoreflect.Message { - mi := &file_plugin_v1_storage_plugin_service_proto_msgTypes[14] + mi := &file_plugin_v1_storage_plugin_service_proto_msgTypes[13] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -818,7 +744,7 @@ func (x *PutObjectResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use PutObjectResponse.ProtoReflect.Descriptor instead. func (*PutObjectResponse) Descriptor() ([]byte, []int) { - return file_plugin_v1_storage_plugin_service_proto_rawDescGZIP(), []int{14} + return file_plugin_v1_storage_plugin_service_proto_rawDescGZIP(), []int{13} } func (x *PutObjectResponse) GetChecksumSha_256() []byte { @@ -848,7 +774,7 @@ type StorageBucketPersisted struct { func (x *StorageBucketPersisted) Reset() { *x = StorageBucketPersisted{} if protoimpl.UnsafeEnabled { - mi := &file_plugin_v1_storage_plugin_service_proto_msgTypes[15] + mi := &file_plugin_v1_storage_plugin_service_proto_msgTypes[14] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -861,7 +787,7 @@ func (x *StorageBucketPersisted) String() string { func (*StorageBucketPersisted) ProtoMessage() {} func (x *StorageBucketPersisted) ProtoReflect() protoreflect.Message { - mi := &file_plugin_v1_storage_plugin_service_proto_msgTypes[15] + mi := &file_plugin_v1_storage_plugin_service_proto_msgTypes[14] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -874,7 +800,7 @@ func (x *StorageBucketPersisted) ProtoReflect() protoreflect.Message { // Deprecated: Use StorageBucketPersisted.ProtoReflect.Descriptor instead. func (*StorageBucketPersisted) Descriptor() ([]byte, []int) { - return file_plugin_v1_storage_plugin_service_proto_rawDescGZIP(), []int{15} + return file_plugin_v1_storage_plugin_service_proto_rawDescGZIP(), []int{14} } func (x *StorageBucketPersisted) GetData() *structpb.Struct { @@ -984,76 +910,70 @@ var file_plugin_v1_storage_plugin_service_proto_rawDesc = []byte{ 0x32, 0x0a, 0x11, 0x47, 0x65, 0x74, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x1d, 0x0a, 0x0a, 0x66, 0x69, 0x6c, 0x65, 0x5f, 0x63, 0x68, 0x75, 0x6e, 0x6b, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x09, 0x66, 0x69, 0x6c, 0x65, 0x43, 0x68, - 0x75, 0x6e, 0x6b, 0x22, 0x78, 0x0a, 0x11, 0x50, 0x75, 0x74, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, - 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x12, 0x51, 0x0a, 0x06, 0x62, 0x75, 0x63, 0x6b, + 0x75, 0x6e, 0x6b, 0x22, 0x8b, 0x01, 0x0a, 0x10, 0x50, 0x75, 0x74, 0x4f, 0x62, 0x6a, 0x65, 0x63, + 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x51, 0x0a, 0x06, 0x62, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x39, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x62, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x52, 0x06, 0x62, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x12, 0x10, 0x0a, 0x03, 0x6b, - 0x65, 0x79, 0x18, 0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x22, 0x75, 0x0a, - 0x10, 0x50, 0x75, 0x74, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, - 0x74, 0x12, 0x1f, 0x0a, 0x0a, 0x66, 0x69, 0x6c, 0x65, 0x5f, 0x63, 0x68, 0x75, 0x6e, 0x6b, 0x18, - 0x0a, 0x20, 0x01, 0x28, 0x0c, 0x48, 0x00, 0x52, 0x09, 0x66, 0x69, 0x6c, 0x65, 0x43, 0x68, 0x75, - 0x6e, 0x6b, 0x12, 0x38, 0x0a, 0x07, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x18, 0x14, 0x20, - 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, - 0x50, 0x75, 0x74, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, - 0x61, 0x48, 0x00, 0x52, 0x07, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x42, 0x06, 0x0a, 0x04, - 0x64, 0x61, 0x74, 0x61, 0x22, 0x3d, 0x0a, 0x11, 0x50, 0x75, 0x74, 0x4f, 0x62, 0x6a, 0x65, 0x63, - 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x28, 0x0a, 0x10, 0x63, 0x68, 0x65, - 0x63, 0x6b, 0x73, 0x75, 0x6d, 0x5f, 0x73, 0x68, 0x61, 0x5f, 0x32, 0x35, 0x36, 0x18, 0x0a, 0x20, - 0x01, 0x28, 0x0c, 0x52, 0x0e, 0x63, 0x68, 0x65, 0x63, 0x6b, 0x73, 0x75, 0x6d, 0x53, 0x68, 0x61, - 0x32, 0x35, 0x36, 0x22, 0x45, 0x0a, 0x16, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, 0x75, - 0x63, 0x6b, 0x65, 0x74, 0x50, 0x65, 0x72, 0x73, 0x69, 0x73, 0x74, 0x65, 0x64, 0x12, 0x2b, 0x0a, - 0x04, 0x64, 0x61, 0x74, 0x61, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x67, 0x6f, - 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x53, 0x74, - 0x72, 0x75, 0x63, 0x74, 0x52, 0x04, 0x64, 0x61, 0x74, 0x61, 0x32, 0x9f, 0x05, 0x0a, 0x14, 0x53, - 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x50, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x53, 0x65, 0x72, 0x76, - 0x69, 0x63, 0x65, 0x12, 0x6a, 0x0a, 0x15, 0x4f, 0x6e, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x53, - 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x12, 0x27, 0x2e, 0x70, - 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x4f, 0x6e, 0x43, 0x72, 0x65, 0x61, 0x74, - 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x52, 0x65, - 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x28, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, - 0x31, 0x2e, 0x4f, 0x6e, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, - 0x65, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, - 0x6a, 0x0a, 0x15, 0x4f, 0x6e, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, - 0x67, 0x65, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x12, 0x27, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, - 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x4f, 0x6e, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x53, 0x74, 0x6f, - 0x72, 0x61, 0x67, 0x65, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, - 0x74, 0x1a, 0x28, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x4f, 0x6e, - 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, 0x75, 0x63, - 0x6b, 0x65, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x6a, 0x0a, 0x15, 0x4f, - 0x6e, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, 0x75, - 0x63, 0x6b, 0x65, 0x74, 0x12, 0x27, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, - 0x2e, 0x4f, 0x6e, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, - 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x28, 0x2e, - 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x4f, 0x6e, 0x44, 0x65, 0x6c, 0x65, - 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x52, - 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x64, 0x0a, 0x13, 0x56, 0x61, 0x6c, 0x69, 0x64, - 0x61, 0x74, 0x65, 0x50, 0x65, 0x72, 0x6d, 0x69, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x25, - 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x56, 0x61, 0x6c, 0x69, 0x64, - 0x61, 0x74, 0x65, 0x50, 0x65, 0x72, 0x6d, 0x69, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65, - 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x26, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, - 0x31, 0x2e, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x50, 0x65, 0x72, 0x6d, 0x69, 0x73, - 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x49, 0x0a, - 0x0a, 0x48, 0x65, 0x61, 0x64, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x12, 0x1c, 0x2e, 0x70, 0x6c, - 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x48, 0x65, 0x61, 0x64, 0x4f, 0x62, 0x6a, 0x65, - 0x63, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1d, 0x2e, 0x70, 0x6c, 0x75, 0x67, - 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x48, 0x65, 0x61, 0x64, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, - 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x48, 0x0a, 0x09, 0x47, 0x65, 0x74, 0x4f, - 0x62, 0x6a, 0x65, 0x63, 0x74, 0x12, 0x1b, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, - 0x31, 0x2e, 0x47, 0x65, 0x74, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, - 0x73, 0x74, 0x1a, 0x1c, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x47, - 0x65, 0x74, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, - 0x30, 0x01, 0x12, 0x48, 0x0a, 0x09, 0x50, 0x75, 0x74, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x12, - 0x1b, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x50, 0x75, 0x74, 0x4f, - 0x62, 0x6a, 0x65, 0x63, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1c, 0x2e, 0x70, + 0x65, 0x79, 0x18, 0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x12, 0x0a, + 0x04, 0x70, 0x61, 0x74, 0x68, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x70, 0x61, 0x74, + 0x68, 0x22, 0x3d, 0x0a, 0x11, 0x50, 0x75, 0x74, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x52, 0x65, + 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x28, 0x0a, 0x10, 0x63, 0x68, 0x65, 0x63, 0x6b, 0x73, + 0x75, 0x6d, 0x5f, 0x73, 0x68, 0x61, 0x5f, 0x32, 0x35, 0x36, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0c, + 0x52, 0x0e, 0x63, 0x68, 0x65, 0x63, 0x6b, 0x73, 0x75, 0x6d, 0x53, 0x68, 0x61, 0x32, 0x35, 0x36, + 0x22, 0x45, 0x0a, 0x16, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, 0x75, 0x63, 0x6b, 0x65, + 0x74, 0x50, 0x65, 0x72, 0x73, 0x69, 0x73, 0x74, 0x65, 0x64, 0x12, 0x2b, 0x0a, 0x04, 0x64, 0x61, + 0x74, 0x61, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, + 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x53, 0x74, 0x72, 0x75, 0x63, + 0x74, 0x52, 0x04, 0x64, 0x61, 0x74, 0x61, 0x32, 0x9d, 0x05, 0x0a, 0x14, 0x53, 0x74, 0x6f, 0x72, + 0x61, 0x67, 0x65, 0x50, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, + 0x12, 0x6a, 0x0a, 0x15, 0x4f, 0x6e, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, + 0x61, 0x67, 0x65, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x12, 0x27, 0x2e, 0x70, 0x6c, 0x75, 0x67, + 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x4f, 0x6e, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x53, 0x74, + 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, + 0x73, 0x74, 0x1a, 0x28, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x4f, + 0x6e, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, 0x75, + 0x63, 0x6b, 0x65, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x6a, 0x0a, 0x15, + 0x4f, 0x6e, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, + 0x75, 0x63, 0x6b, 0x65, 0x74, 0x12, 0x27, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, + 0x31, 0x2e, 0x4f, 0x6e, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, + 0x65, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x28, + 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x4f, 0x6e, 0x55, 0x70, 0x64, + 0x61, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, + 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x6a, 0x0a, 0x15, 0x4f, 0x6e, 0x44, 0x65, + 0x6c, 0x65, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, 0x75, 0x63, 0x6b, 0x65, + 0x74, 0x12, 0x27, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x4f, 0x6e, + 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x53, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, 0x75, 0x63, + 0x6b, 0x65, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x28, 0x2e, 0x70, 0x6c, 0x75, + 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x4f, 0x6e, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x53, + 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x52, 0x65, 0x73, 0x70, + 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x64, 0x0a, 0x13, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, + 0x50, 0x65, 0x72, 0x6d, 0x69, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x25, 0x2e, 0x70, 0x6c, + 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, + 0x50, 0x65, 0x72, 0x6d, 0x69, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, + 0x73, 0x74, 0x1a, 0x26, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x56, + 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x50, 0x65, 0x72, 0x6d, 0x69, 0x73, 0x73, 0x69, 0x6f, + 0x6e, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x49, 0x0a, 0x0a, 0x48, 0x65, + 0x61, 0x64, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x12, 0x1c, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, + 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x48, 0x65, 0x61, 0x64, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x52, + 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1d, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, + 0x76, 0x31, 0x2e, 0x48, 0x65, 0x61, 0x64, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x52, 0x65, 0x73, + 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x48, 0x0a, 0x09, 0x47, 0x65, 0x74, 0x4f, 0x62, 0x6a, 0x65, + 0x63, 0x74, 0x12, 0x1b, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x47, + 0x65, 0x74, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, + 0x1c, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x47, 0x65, 0x74, 0x4f, + 0x62, 0x6a, 0x65, 0x63, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x30, 0x01, 0x12, + 0x46, 0x0a, 0x09, 0x50, 0x75, 0x74, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x12, 0x1b, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x50, 0x75, 0x74, 0x4f, 0x62, 0x6a, 0x65, - 0x63, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x28, 0x01, 0x42, 0x35, 0x5a, 0x33, - 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, - 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, 0x73, 0x64, - 0x6b, 0x2f, 0x70, 0x62, 0x73, 0x2f, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x3b, 0x70, 0x6c, 0x75, - 0x67, 0x69, 0x6e, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x63, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1c, 0x2e, 0x70, 0x6c, 0x75, 0x67, + 0x69, 0x6e, 0x2e, 0x76, 0x31, 0x2e, 0x50, 0x75, 0x74, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x52, + 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x35, 0x5a, 0x33, 0x67, 0x69, 0x74, 0x68, 0x75, + 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, + 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, 0x73, 0x64, 0x6b, 0x2f, 0x70, 0x62, 0x73, + 0x2f, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x3b, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x62, 0x06, + 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( @@ -1068,7 +988,7 @@ func file_plugin_v1_storage_plugin_service_proto_rawDescGZIP() []byte { return file_plugin_v1_storage_plugin_service_proto_rawDescData } -var file_plugin_v1_storage_plugin_service_proto_msgTypes = make([]protoimpl.MessageInfo, 16) +var file_plugin_v1_storage_plugin_service_proto_msgTypes = make([]protoimpl.MessageInfo, 15) var file_plugin_v1_storage_plugin_service_proto_goTypes = []interface{}{ (*OnCreateStorageBucketRequest)(nil), // 0: plugin.v1.OnCreateStorageBucketRequest (*OnCreateStorageBucketResponse)(nil), // 1: plugin.v1.OnCreateStorageBucketResponse @@ -1082,49 +1002,47 @@ var file_plugin_v1_storage_plugin_service_proto_goTypes = []interface{}{ (*HeadObjectResponse)(nil), // 9: plugin.v1.HeadObjectResponse (*GetObjectRequest)(nil), // 10: plugin.v1.GetObjectRequest (*GetObjectResponse)(nil), // 11: plugin.v1.GetObjectResponse - (*PutObjectMetadata)(nil), // 12: plugin.v1.PutObjectMetadata - (*PutObjectRequest)(nil), // 13: plugin.v1.PutObjectRequest - (*PutObjectResponse)(nil), // 14: plugin.v1.PutObjectResponse - (*StorageBucketPersisted)(nil), // 15: plugin.v1.StorageBucketPersisted - (*storagebuckets.StorageBucket)(nil), // 16: controller.api.resources.storagebuckets.v1.StorageBucket - (*timestamppb.Timestamp)(nil), // 17: google.protobuf.Timestamp - (*structpb.Struct)(nil), // 18: google.protobuf.Struct + (*PutObjectRequest)(nil), // 12: plugin.v1.PutObjectRequest + (*PutObjectResponse)(nil), // 13: plugin.v1.PutObjectResponse + (*StorageBucketPersisted)(nil), // 14: plugin.v1.StorageBucketPersisted + (*storagebuckets.StorageBucket)(nil), // 15: controller.api.resources.storagebuckets.v1.StorageBucket + (*timestamppb.Timestamp)(nil), // 16: google.protobuf.Timestamp + (*structpb.Struct)(nil), // 17: google.protobuf.Struct } var file_plugin_v1_storage_plugin_service_proto_depIdxs = []int32{ - 16, // 0: plugin.v1.OnCreateStorageBucketRequest.bucket:type_name -> controller.api.resources.storagebuckets.v1.StorageBucket - 15, // 1: plugin.v1.OnCreateStorageBucketResponse.persisted:type_name -> plugin.v1.StorageBucketPersisted - 16, // 2: plugin.v1.OnUpdateStorageBucketRequest.current_bucket:type_name -> controller.api.resources.storagebuckets.v1.StorageBucket - 16, // 3: plugin.v1.OnUpdateStorageBucketRequest.new_bucket:type_name -> controller.api.resources.storagebuckets.v1.StorageBucket - 15, // 4: plugin.v1.OnUpdateStorageBucketRequest.persisted:type_name -> plugin.v1.StorageBucketPersisted - 15, // 5: plugin.v1.OnUpdateStorageBucketResponse.persisted:type_name -> plugin.v1.StorageBucketPersisted - 16, // 6: plugin.v1.OnDeleteStorageBucketRequest.bucket:type_name -> controller.api.resources.storagebuckets.v1.StorageBucket - 15, // 7: plugin.v1.OnDeleteStorageBucketRequest.persisted:type_name -> plugin.v1.StorageBucketPersisted - 16, // 8: plugin.v1.ValidatePermissionsRequest.bucket:type_name -> controller.api.resources.storagebuckets.v1.StorageBucket - 16, // 9: plugin.v1.HeadObjectRequest.bucket:type_name -> controller.api.resources.storagebuckets.v1.StorageBucket - 17, // 10: plugin.v1.HeadObjectResponse.last_modified:type_name -> google.protobuf.Timestamp - 16, // 11: plugin.v1.GetObjectRequest.bucket:type_name -> controller.api.resources.storagebuckets.v1.StorageBucket - 16, // 12: plugin.v1.PutObjectMetadata.bucket:type_name -> controller.api.resources.storagebuckets.v1.StorageBucket - 12, // 13: plugin.v1.PutObjectRequest.request:type_name -> plugin.v1.PutObjectMetadata - 18, // 14: plugin.v1.StorageBucketPersisted.data:type_name -> google.protobuf.Struct - 0, // 15: plugin.v1.StoragePluginService.OnCreateStorageBucket:input_type -> plugin.v1.OnCreateStorageBucketRequest - 2, // 16: plugin.v1.StoragePluginService.OnUpdateStorageBucket:input_type -> plugin.v1.OnUpdateStorageBucketRequest - 4, // 17: plugin.v1.StoragePluginService.OnDeleteStorageBucket:input_type -> plugin.v1.OnDeleteStorageBucketRequest - 6, // 18: plugin.v1.StoragePluginService.ValidatePermissions:input_type -> plugin.v1.ValidatePermissionsRequest - 8, // 19: plugin.v1.StoragePluginService.HeadObject:input_type -> plugin.v1.HeadObjectRequest - 10, // 20: plugin.v1.StoragePluginService.GetObject:input_type -> plugin.v1.GetObjectRequest - 13, // 21: plugin.v1.StoragePluginService.PutObject:input_type -> plugin.v1.PutObjectRequest - 1, // 22: plugin.v1.StoragePluginService.OnCreateStorageBucket:output_type -> plugin.v1.OnCreateStorageBucketResponse - 3, // 23: plugin.v1.StoragePluginService.OnUpdateStorageBucket:output_type -> plugin.v1.OnUpdateStorageBucketResponse - 5, // 24: plugin.v1.StoragePluginService.OnDeleteStorageBucket:output_type -> plugin.v1.OnDeleteStorageBucketResponse - 7, // 25: plugin.v1.StoragePluginService.ValidatePermissions:output_type -> plugin.v1.ValidatePermissionsResponse - 9, // 26: plugin.v1.StoragePluginService.HeadObject:output_type -> plugin.v1.HeadObjectResponse - 11, // 27: plugin.v1.StoragePluginService.GetObject:output_type -> plugin.v1.GetObjectResponse - 14, // 28: plugin.v1.StoragePluginService.PutObject:output_type -> plugin.v1.PutObjectResponse - 22, // [22:29] is the sub-list for method output_type - 15, // [15:22] is the sub-list for method input_type - 15, // [15:15] is the sub-list for extension type_name - 15, // [15:15] is the sub-list for extension extendee - 0, // [0:15] is the sub-list for field type_name + 15, // 0: plugin.v1.OnCreateStorageBucketRequest.bucket:type_name -> controller.api.resources.storagebuckets.v1.StorageBucket + 14, // 1: plugin.v1.OnCreateStorageBucketResponse.persisted:type_name -> plugin.v1.StorageBucketPersisted + 15, // 2: plugin.v1.OnUpdateStorageBucketRequest.current_bucket:type_name -> controller.api.resources.storagebuckets.v1.StorageBucket + 15, // 3: plugin.v1.OnUpdateStorageBucketRequest.new_bucket:type_name -> controller.api.resources.storagebuckets.v1.StorageBucket + 14, // 4: plugin.v1.OnUpdateStorageBucketRequest.persisted:type_name -> plugin.v1.StorageBucketPersisted + 14, // 5: plugin.v1.OnUpdateStorageBucketResponse.persisted:type_name -> plugin.v1.StorageBucketPersisted + 15, // 6: plugin.v1.OnDeleteStorageBucketRequest.bucket:type_name -> controller.api.resources.storagebuckets.v1.StorageBucket + 14, // 7: plugin.v1.OnDeleteStorageBucketRequest.persisted:type_name -> plugin.v1.StorageBucketPersisted + 15, // 8: plugin.v1.ValidatePermissionsRequest.bucket:type_name -> controller.api.resources.storagebuckets.v1.StorageBucket + 15, // 9: plugin.v1.HeadObjectRequest.bucket:type_name -> controller.api.resources.storagebuckets.v1.StorageBucket + 16, // 10: plugin.v1.HeadObjectResponse.last_modified:type_name -> google.protobuf.Timestamp + 15, // 11: plugin.v1.GetObjectRequest.bucket:type_name -> controller.api.resources.storagebuckets.v1.StorageBucket + 15, // 12: plugin.v1.PutObjectRequest.bucket:type_name -> controller.api.resources.storagebuckets.v1.StorageBucket + 17, // 13: plugin.v1.StorageBucketPersisted.data:type_name -> google.protobuf.Struct + 0, // 14: plugin.v1.StoragePluginService.OnCreateStorageBucket:input_type -> plugin.v1.OnCreateStorageBucketRequest + 2, // 15: plugin.v1.StoragePluginService.OnUpdateStorageBucket:input_type -> plugin.v1.OnUpdateStorageBucketRequest + 4, // 16: plugin.v1.StoragePluginService.OnDeleteStorageBucket:input_type -> plugin.v1.OnDeleteStorageBucketRequest + 6, // 17: plugin.v1.StoragePluginService.ValidatePermissions:input_type -> plugin.v1.ValidatePermissionsRequest + 8, // 18: plugin.v1.StoragePluginService.HeadObject:input_type -> plugin.v1.HeadObjectRequest + 10, // 19: plugin.v1.StoragePluginService.GetObject:input_type -> plugin.v1.GetObjectRequest + 12, // 20: plugin.v1.StoragePluginService.PutObject:input_type -> plugin.v1.PutObjectRequest + 1, // 21: plugin.v1.StoragePluginService.OnCreateStorageBucket:output_type -> plugin.v1.OnCreateStorageBucketResponse + 3, // 22: plugin.v1.StoragePluginService.OnUpdateStorageBucket:output_type -> plugin.v1.OnUpdateStorageBucketResponse + 5, // 23: plugin.v1.StoragePluginService.OnDeleteStorageBucket:output_type -> plugin.v1.OnDeleteStorageBucketResponse + 7, // 24: plugin.v1.StoragePluginService.ValidatePermissions:output_type -> plugin.v1.ValidatePermissionsResponse + 9, // 25: plugin.v1.StoragePluginService.HeadObject:output_type -> plugin.v1.HeadObjectResponse + 11, // 26: plugin.v1.StoragePluginService.GetObject:output_type -> plugin.v1.GetObjectResponse + 13, // 27: plugin.v1.StoragePluginService.PutObject:output_type -> plugin.v1.PutObjectResponse + 21, // [21:28] is the sub-list for method output_type + 14, // [14:21] is the sub-list for method input_type + 14, // [14:14] is the sub-list for extension type_name + 14, // [14:14] is the sub-list for extension extendee + 0, // [0:14] is the sub-list for field type_name } func init() { file_plugin_v1_storage_plugin_service_proto_init() } @@ -1278,18 +1196,6 @@ func file_plugin_v1_storage_plugin_service_proto_init() { } } file_plugin_v1_storage_plugin_service_proto_msgTypes[12].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*PutObjectMetadata); i { - case 0: - return &v.state - case 1: - return &v.sizeCache - case 2: - return &v.unknownFields - default: - return nil - } - } - file_plugin_v1_storage_plugin_service_proto_msgTypes[13].Exporter = func(v interface{}, i int) interface{} { switch v := v.(*PutObjectRequest); i { case 0: return &v.state @@ -1301,7 +1207,7 @@ func file_plugin_v1_storage_plugin_service_proto_init() { return nil } } - file_plugin_v1_storage_plugin_service_proto_msgTypes[14].Exporter = func(v interface{}, i int) interface{} { + file_plugin_v1_storage_plugin_service_proto_msgTypes[13].Exporter = func(v interface{}, i int) interface{} { switch v := v.(*PutObjectResponse); i { case 0: return &v.state @@ -1313,7 +1219,7 @@ func file_plugin_v1_storage_plugin_service_proto_init() { return nil } } - file_plugin_v1_storage_plugin_service_proto_msgTypes[15].Exporter = func(v interface{}, i int) interface{} { + file_plugin_v1_storage_plugin_service_proto_msgTypes[14].Exporter = func(v interface{}, i int) interface{} { switch v := v.(*StorageBucketPersisted); i { case 0: return &v.state @@ -1326,17 +1232,13 @@ func file_plugin_v1_storage_plugin_service_proto_init() { } } } - file_plugin_v1_storage_plugin_service_proto_msgTypes[13].OneofWrappers = []interface{}{ - (*PutObjectRequest_FileChunk)(nil), - (*PutObjectRequest_Request)(nil), - } type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: file_plugin_v1_storage_plugin_service_proto_rawDesc, NumEnums: 0, - NumMessages: 16, + NumMessages: 15, NumExtensions: 0, NumServices: 1, }, diff --git a/sdk/pbs/plugin/storage_plugin_service_grpc.pb.go b/sdk/pbs/plugin/storage_plugin_service_grpc.pb.go index 862d26d63..2f75c16e9 100644 --- a/sdk/pbs/plugin/storage_plugin_service_grpc.pb.go +++ b/sdk/pbs/plugin/storage_plugin_service_grpc.pb.go @@ -34,9 +34,9 @@ type StoragePluginServiceClient interface { HeadObject(ctx context.Context, in *HeadObjectRequest, opts ...grpc.CallOption) (*HeadObjectResponse, error) // GetObject is a hook that retrieves objects. GetObject(ctx context.Context, in *GetObjectRequest, opts ...grpc.CallOption) (StoragePluginService_GetObjectClient, error) - // PutObject is a hook that streams chunks of a file to be stored - // in an external object store. - PutObject(ctx context.Context, opts ...grpc.CallOption) (StoragePluginService_PutObjectClient, error) + // PutObject is a hook that reads a file stored on local disk and + // stores it to an external object store. + PutObject(ctx context.Context, in *PutObjectRequest, opts ...grpc.CallOption) (*PutObjectResponse, error) } type storagePluginServiceClient struct { @@ -124,38 +124,13 @@ func (x *storagePluginServiceGetObjectClient) Recv() (*GetObjectResponse, error) return m, nil } -func (c *storagePluginServiceClient) PutObject(ctx context.Context, opts ...grpc.CallOption) (StoragePluginService_PutObjectClient, error) { - stream, err := c.cc.NewStream(ctx, &StoragePluginService_ServiceDesc.Streams[1], "/plugin.v1.StoragePluginService/PutObject", opts...) +func (c *storagePluginServiceClient) PutObject(ctx context.Context, in *PutObjectRequest, opts ...grpc.CallOption) (*PutObjectResponse, error) { + out := new(PutObjectResponse) + err := c.cc.Invoke(ctx, "/plugin.v1.StoragePluginService/PutObject", in, out, opts...) if err != nil { return nil, err } - x := &storagePluginServicePutObjectClient{stream} - return x, nil -} - -type StoragePluginService_PutObjectClient interface { - Send(*PutObjectRequest) error - CloseAndRecv() (*PutObjectResponse, error) - grpc.ClientStream -} - -type storagePluginServicePutObjectClient struct { - grpc.ClientStream -} - -func (x *storagePluginServicePutObjectClient) Send(m *PutObjectRequest) error { - return x.ClientStream.SendMsg(m) -} - -func (x *storagePluginServicePutObjectClient) CloseAndRecv() (*PutObjectResponse, error) { - if err := x.ClientStream.CloseSend(); err != nil { - return nil, err - } - m := new(PutObjectResponse) - if err := x.ClientStream.RecvMsg(m); err != nil { - return nil, err - } - return m, nil + return out, nil } // StoragePluginServiceServer is the server API for StoragePluginService service. @@ -178,9 +153,9 @@ type StoragePluginServiceServer interface { HeadObject(context.Context, *HeadObjectRequest) (*HeadObjectResponse, error) // GetObject is a hook that retrieves objects. GetObject(*GetObjectRequest, StoragePluginService_GetObjectServer) error - // PutObject is a hook that streams chunks of a file to be stored - // in an external object store. - PutObject(StoragePluginService_PutObjectServer) error + // PutObject is a hook that reads a file stored on local disk and + // stores it to an external object store. + PutObject(context.Context, *PutObjectRequest) (*PutObjectResponse, error) mustEmbedUnimplementedStoragePluginServiceServer() } @@ -206,8 +181,8 @@ func (UnimplementedStoragePluginServiceServer) HeadObject(context.Context, *Head func (UnimplementedStoragePluginServiceServer) GetObject(*GetObjectRequest, StoragePluginService_GetObjectServer) error { return status.Errorf(codes.Unimplemented, "method GetObject not implemented") } -func (UnimplementedStoragePluginServiceServer) PutObject(StoragePluginService_PutObjectServer) error { - return status.Errorf(codes.Unimplemented, "method PutObject not implemented") +func (UnimplementedStoragePluginServiceServer) PutObject(context.Context, *PutObjectRequest) (*PutObjectResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method PutObject not implemented") } func (UnimplementedStoragePluginServiceServer) mustEmbedUnimplementedStoragePluginServiceServer() {} @@ -333,30 +308,22 @@ func (x *storagePluginServiceGetObjectServer) Send(m *GetObjectResponse) error { return x.ServerStream.SendMsg(m) } -func _StoragePluginService_PutObject_Handler(srv interface{}, stream grpc.ServerStream) error { - return srv.(StoragePluginServiceServer).PutObject(&storagePluginServicePutObjectServer{stream}) -} - -type StoragePluginService_PutObjectServer interface { - SendAndClose(*PutObjectResponse) error - Recv() (*PutObjectRequest, error) - grpc.ServerStream -} - -type storagePluginServicePutObjectServer struct { - grpc.ServerStream -} - -func (x *storagePluginServicePutObjectServer) SendAndClose(m *PutObjectResponse) error { - return x.ServerStream.SendMsg(m) -} - -func (x *storagePluginServicePutObjectServer) Recv() (*PutObjectRequest, error) { - m := new(PutObjectRequest) - if err := x.ServerStream.RecvMsg(m); err != nil { +func _StoragePluginService_PutObject_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(PutObjectRequest) + if err := dec(in); err != nil { return nil, err } - return m, nil + if interceptor == nil { + return srv.(StoragePluginServiceServer).PutObject(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/plugin.v1.StoragePluginService/PutObject", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(StoragePluginServiceServer).PutObject(ctx, req.(*PutObjectRequest)) + } + return interceptor(ctx, in, info, handler) } // StoragePluginService_ServiceDesc is the grpc.ServiceDesc for StoragePluginService service. @@ -386,6 +353,10 @@ var StoragePluginService_ServiceDesc = grpc.ServiceDesc{ MethodName: "HeadObject", Handler: _StoragePluginService_HeadObject_Handler, }, + { + MethodName: "PutObject", + Handler: _StoragePluginService_PutObject_Handler, + }, }, Streams: []grpc.StreamDesc{ { @@ -393,11 +364,6 @@ var StoragePluginService_ServiceDesc = grpc.ServiceDesc{ Handler: _StoragePluginService_GetObject_Handler, ServerStreams: true, }, - { - StreamName: "PutObject", - Handler: _StoragePluginService_PutObject_Handler, - ClientStreams: true, - }, }, Metadata: "plugin/v1/storage_plugin_service.proto", }