gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(bssdk): private createBlob and sealBlob

Browse Source 
with: #22918200

Signed-off-by: mawei029 <mawei2@oppo.com>
This commit is contained in:
mawei029 2024-12-24 16:49:41 +08:00 committed by slasher
parent e20adade26
commit 94e068fa65
3 changed files with 37 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
blobstore/api/access/client.go
View File

@ -217,9 +217,7 @@ type API interface {
type Client interface {
API
CreateBlob(ctx context.Context, args *CreateBlobArgs) (CreateBlobRet, error)
ListBlob(ctx context.Context, args *ListBlobArgs) (shardnode.ListBlobRet, error)
SealBlob(ctx context.Context, args *SealBlobArgs) error
GetBlob(ctx context.Context, args *GetBlobArgs) (io.ReadCloser, error)
DeleteBlob(ctx context.Context, args *DelBlobArgs) error
PutBlob(ctx context.Context, args *PutBlobArgs) (proto.ClusterID, HashSumMap, error)

58
blobstore/sdk/sdk_client.go
View File

@ -239,22 +239,6 @@ func (s *sdkHandler) ListBlob(ctx context.Context, args *acapi.ListBlobArgs) (sh
return s.handler.ListBlob(ctx, args)
}
func (s *sdkHandler) CreateBlob(ctx context.Context, args *acapi.CreateBlobArgs) (acapi.CreateBlobRet, error) {
if !args.IsValid() {
return acapi.CreateBlobRet{}, errcode.ErrIllegalArguments
}
ctx = acapi.ClientWithReqidContext(ctx)
span := trace.SpanFromContextSafe(ctx)
span.Debugf("accept sdk CreateBlob request, name=%s, keys=%s, args: %v", args.BlobName, args.ShardKeys, *args)
loc, err := s.handler.CreateBlob(ctx, args)
if err != nil {
return acapi.CreateBlobRet{}, err
}
return acapi.CreateBlobRet{Location: *loc}, nil
}
func (s *sdkHandler) DeleteBlob(ctx context.Context, args *acapi.DelBlobArgs) error {
if !args.IsValid() {
return errcode.ErrIllegalArguments
@ -267,17 +251,6 @@ func (s *sdkHandler) DeleteBlob(ctx context.Context, args *acapi.DelBlobArgs) er
return s.handler.DeleteBlob(ctx, args)
}
func (s *sdkHandler) SealBlob(ctx context.Context, args *acapi.SealBlobArgs) error {
if !args.IsValid() {
return errcode.ErrIllegalArguments
}
ctx = acapi.ClientWithReqidContext(ctx)
span := trace.SpanFromContextSafe(ctx)
span.Debugf("accept sdk SealBlob request, name=%s, keys=%s, args: %v", args.BlobName, args.ShardKeys, *args)
return s.handler.SealBlob(ctx, args)
}
func (s *sdkHandler) GetBlob(ctx context.Context, args *acapi.GetBlobArgs) (io.ReadCloser, error) {
if !args.IsValid() {
return nil, errcode.ErrIllegalArguments
@ -348,7 +321,7 @@ func (s *sdkHandler) PutBlob(ctx context.Context, args *acapi.PutBlobArgs) (cid
ClusterID: loc.ClusterID,
Slices: loc.Slices,
}
if err = s.SealBlob(ctx, sealArgs); err != nil {
if err = s.sealBlob(ctx, sealArgs); err != nil {
span.Warnf("seal fail, seal args=%v", sealArgs)
return loc.ClusterID, nil, err
}
@ -356,6 +329,33 @@ func (s *sdkHandler) PutBlob(ctx context.Context, args *acapi.PutBlobArgs) (cid
return loc.ClusterID, hashes, nil
}
func (s *sdkHandler) createBlob(ctx context.Context, args *acapi.CreateBlobArgs) (acapi.CreateBlobRet, error) {
if !args.IsValid() {
return acapi.CreateBlobRet{}, errcode.ErrIllegalArguments
}
ctx = acapi.ClientWithReqidContext(ctx)
span := trace.SpanFromContextSafe(ctx)
span.Debugf("accept sdk CreateBlob request, name=%s, keys=%s, args: %v", args.BlobName, args.ShardKeys, *args)
loc, err := s.handler.CreateBlob(ctx, args)
if err != nil {
return acapi.CreateBlobRet{}, err
}
return acapi.CreateBlobRet{Location: *loc}, nil
}
func (s *sdkHandler) sealBlob(ctx context.Context, args *acapi.SealBlobArgs) error {
if !args.IsValid() {
return errcode.ErrIllegalArguments
}
ctx = acapi.ClientWithReqidContext(ctx)
span := trace.SpanFromContextSafe(ctx)
span.Debugf("accept sdk SealBlob request, name=%s, keys=%s, args: %v", args.BlobName, args.ShardKeys, *args)
return s.handler.SealBlob(ctx, args)
}
func (s *sdkHandler) alloc(ctx context.Context, args *acapi.AllocArgs) (acapi.AllocResp, error) {
if !args.IsValid() {
return acapi.AllocResp{}, errcode.ErrIllegalArguments
@ -932,7 +932,7 @@ func (s *sdkHandler) putParts(ctx context.Context, args *acapi.PutArgs) (proto.L
func (s *sdkHandler) putBlobs(ctx context.Context, args *acapi.PutBlobArgs) (proto.Location, acapi.HashSumMap, error) {
// create
created, err := s.CreateBlob(ctx, &acapi.CreateBlobArgs{
created, err := s.createBlob(ctx, &acapi.CreateBlobArgs{
BlobName: args.BlobName,
ShardKeys: args.ShardKeys,
CodeMode: args.CodeMode,

16
blobstore/sdk/sdk_client_test.go
View File

@ -505,12 +505,12 @@ func TestSdkBlob_Create(t *testing.T) {
hd := newSdkHandler(t)
hd.conf.ShardnodeConfig = &stream.ShardnodeConfig{}
_, err := hd.CreateBlob(ctx, nil)
_, err := hd.createBlob(ctx, nil)
require.NotNil(t, err)
require.ErrorIs(t, err, errcode.ErrIllegalArguments)
args := &acapi.CreateBlobArgs{}
_, err = hd.CreateBlob(ctx, args)
_, err = hd.createBlob(ctx, args)
require.NotNil(t, err)
require.ErrorIs(t, err, errcode.ErrIllegalArguments)
@ -518,7 +518,7 @@ func TestSdkBlob_Create(t *testing.T) {
args.CodeMode = codemode.EC3P3
args.Size = 1
args.BlobName = []byte("blob1")
_, err = hd.CreateBlob(ctx, args)
_, err = hd.createBlob(ctx, args)
require.NotNil(t, err)
require.ErrorIs(t, err, errMock)
@ -528,7 +528,7 @@ func TestSdkBlob_Create(t *testing.T) {
SliceSize: 1,
}
hd.handler.(*mocks.MockStreamHandler).EXPECT().CreateBlob(gAny, gAny).Return(loca, nil)
_, err = hd.CreateBlob(ctx, args)
_, err = hd.createBlob(ctx, args)
require.NoError(t, err)
}
@ -537,12 +537,12 @@ func TestSdkBlob_Seal(t *testing.T) {
hd := newSdkHandler(t)
hd.conf.ShardnodeConfig = &stream.ShardnodeConfig{}
err := hd.SealBlob(ctx, nil)
err := hd.sealBlob(ctx, nil)
require.NotNil(t, err)
require.ErrorIs(t, err, errcode.ErrIllegalArguments)
args := &acapi.SealBlobArgs{}
err = hd.SealBlob(ctx, args)
err = hd.sealBlob(ctx, args)
require.NotNil(t, err)
require.ErrorIs(t, err, errcode.ErrIllegalArguments)
@ -550,12 +550,12 @@ func TestSdkBlob_Seal(t *testing.T) {
args.ClusterID = 1
args.BlobName = []byte("blob1")
args.Slices = make([]proto.Slice, 1)
err = hd.SealBlob(ctx, args)
err = hd.sealBlob(ctx, args)
require.NotNil(t, err)
require.ErrorIs(t, err, errMock)
hd.handler.(*mocks.MockStreamHandler).EXPECT().SealBlob(gAny, gAny).Return(nil)
err = hd.SealBlob(ctx, args)
err = hd.sealBlob(ctx, args)
require.NoError(t, err)
}