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

feat(bssdk): support specify code_mode and cluster_id for access sdk put api

Browse Source 
fixes: https://github.com/cubefs/cubefs/issues/3954

@formatter:off

Signed-off-by: zhangjianwei2 <zhangjianwei2@cmss.chinamobile.com>
This commit is contained in:
zhangjianwei2 2025-11-17 10:06:29 +08:00 committed by 梁曟風
parent ed09c33f3b
commit 00fa06c716
18 changed files with 275 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
blobstore/access/server.go
View File

@ -234,7 +234,7 @@ func (s *Service) Put(c *rpc.Context) {
}
rc := s.limiter.Reader(ctx, c.Request.Body)
loc, err := s.streamHandler.Put(ctx, rc, args.Size, hasherMap)
loc, err := s.streamHandler.Put(ctx, rc, args.Size, hasherMap, args.AssignClusterID, args.CodeMode)
if err != nil {
span.Error("stream put failed", errors.Detail(err))
c.RespondError(httpError(err))

101
blobstore/access/server_test.go
View File

@ -46,7 +46,7 @@ var (
_blobSize uint32 = 1 << 20
location = &proto.Location{
ClusterID: 1,
CodeMode: 1,
CodeMode: codemode.EC15P12,
SliceSize: _blobSize,
Crc: 0,
Slices: []proto.Slice{{
@ -56,6 +56,42 @@ var (
}},
}
locationForClusterID = &proto.Location{
ClusterID: 11,
CodeMode: codemode.EC3P3,
SliceSize: _blobSize,
Crc: 0,
Slices: []proto.Slice{{
MinSliceID: 111,
Vid: 1111,
Count: 1,
}},
}
locationForCodeMode = &proto.Location{
ClusterID: 22,
CodeMode: codemode.EC6P6,
SliceSize: _blobSize,
Crc: 0,
Slices: []proto.Slice{{
MinSliceID: 111,
Vid: 1111,
Count: 1,
}},
}
locationForClusterIDAndCodeMode = &proto.Location{
ClusterID: 99,
CodeMode: codemode.EC12P4,
SliceSize: _blobSize,
Crc: 0,
Slices: []proto.Slice{{
MinSliceID: 111,
Vid: 1111,
Count: 1,
}},
}
testServer *httptest.Server
once sync.Once
)
@ -95,12 +131,23 @@ func newService() *Service {
return nil
})
s.EXPECT().Put(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(
func(ctx context.Context, rc io.Reader, size int64, hasherMap access.HasherMap) (*proto.Location, error) {
s.EXPECT().Put(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(
func(ctx context.Context, rc io.Reader, size int64, hasherMap access.HasherMap,
assignClusterID proto.ClusterID, codeMode codemode.CodeMode,
) (*proto.Location, error) {
if size < 1024 {
return nil, errors.New("fake put nil body")
}
loc := location.Copy()
var loc proto.Location
if assignClusterID == 0 && codeMode == codemode.CodeModeNone {
loc = location.Copy()
} else if assignClusterID == 11 && codeMode == codemode.CodeModeNone {
loc = locationForClusterID.Copy()
} else if assignClusterID == 0 && codeMode == codemode.EC6P6 {
loc = locationForCodeMode.Copy()
} else if assignClusterID == 99 && codeMode == codemode.EC12P4 {
loc = locationForClusterIDAndCodeMode.Copy()
}
loc.Size_ = uint64(size)
security.LocationCrcFill(&loc)
return &loc, nil
@ -249,6 +296,21 @@ func TestAccessServicePut(t *testing.T) {
return fmt.Sprintf("%s/put?size=%d&hashes=%d", host, size, hashes)
}
urlForClusterID := func(size int64, hashes access.HashAlgorithm, assignClusterID proto.ClusterID) string {
return fmt.Sprintf("%s/put?size=%d&hashes=%d&assign_cluster_id=%d", host, size, hashes, assignClusterID)
}
urlForCodeMode := func(size int64, hashes access.HashAlgorithm, codeMode codemode.CodeMode) string {
return fmt.Sprintf("%s/put?size=%d&hashes=%d&code_mode=%d", host, size, hashes, codeMode)
}
urlForClusterIDAndCodeMode := func(size int64, hashes access.HashAlgorithm,
assignClusterID proto.ClusterID, codeMode codemode.CodeMode,
) string {
return fmt.Sprintf("%s/put?size=%d&hashes=%d&assign_cluster_id=%d&code_mode=%d",
host, size, hashes, assignClusterID, codeMode)
}
for _, method := range []string{http.MethodPut, http.MethodPost} {
args := access.PutArgs{
Size: 0,
@ -281,6 +343,37 @@ func TestAccessServicePut(t *testing.T) {
err := cli.DoWith(ctx, req, resp, rpc.WithCrcEncode())
require.NoError(t, err)
require.Equal(t, uint64(1024), resp.Location.Size_)
require.Equal(t, proto.ClusterID(1), resp.Location.ClusterID)
require.Equal(t, codemode.EC15P12, resp.Location.CodeMode)
}
{
args.Body = bytes.NewReader(make([]byte, 1024))
req, _ := http.NewRequest(method,
urlForClusterIDAndCodeMode(1024, args.Hashes, proto.ClusterID(99), codemode.EC12P4), args.Body)
resp := &access.PutResp{}
err := cli.DoWith(ctx, req, resp, rpc.WithCrcEncode())
require.NoError(t, err)
require.Equal(t, uint64(1024), resp.Location.Size_)
require.Equal(t, proto.ClusterID(99), resp.Location.ClusterID)
require.Equal(t, codemode.EC12P4, resp.Location.CodeMode)
}
{
args.Body = bytes.NewReader(make([]byte, 1024))
req, _ := http.NewRequest(method, urlForClusterID(1024, args.Hashes, proto.ClusterID(11)), args.Body)
resp := &access.PutResp{}
err := cli.DoWith(ctx, req, resp, rpc.WithCrcEncode())
require.NoError(t, err)
require.Equal(t, uint64(1024), resp.Location.Size_)
require.Equal(t, proto.ClusterID(11), resp.Location.ClusterID)
}
{
args.Body = bytes.NewReader(make([]byte, 1024))
req, _ := http.NewRequest(method, urlForCodeMode(1024, args.Hashes, codemode.EC6P6), args.Body)
resp := &access.PutResp{}
err := cli.DoWith(ctx, req, resp, rpc.WithCrcEncode())
require.NoError(t, err)
require.Equal(t, uint64(1024), resp.Location.Size_)
require.Equal(t, codemode.EC6P6, resp.Location.CodeMode)
}
}
}

19
blobstore/access/stream/codemode.go
View File

@ -43,3 +43,22 @@ func (c CodeModePairs) SelectCodeMode(size int64) codemode.CodeMode {
panic(fmt.Sprintf("no codemode policy to be selected by size %d, %+v", size, c))
}
// Verify select codemode
func (c CodeModePairs) VerifySelectCodeMode(selectCodeMode codemode.CodeMode) bool {
if !selectCodeMode.IsValid() {
return false
}
for codeMode, pair := range c {
policy := pair.Policy
if !policy.Enable {
continue
}
if selectCodeMode == codeMode {
return true
}
}
return false
}

4
blobstore/access/stream/stream.go
View File

@ -74,7 +74,9 @@ type StreamHandler interface {
// Put put one object
// required: size, file size
// optional: hasher map to calculate hash.Hash
Put(ctx context.Context, rc io.Reader, size int64, hasherMap access.HasherMap) (*proto.Location, error)
// optional: code to specify codemode and not choose codemode by size
Put(ctx context.Context, rc io.Reader, size int64, hasherMap access.HasherMap,
assignClusterID proto.ClusterID, codeMode codemode.CodeMode) (*proto.Location, error)
// Get read file
// required: location, readSize

2
blobstore/access/stream/stream_alloc.go
View File

@ -57,7 +57,7 @@ func (h *Handler) Alloc(ctx context.Context, size uint64, blobSize uint32,
span.Debugf("fill blobsize:%d", blobSize)
}
if codeMode == 0 {
if codeMode == codemode.CodeModeNone {
codeMode = h.allCodeModes.SelectCodeMode(int64(size))
span.Debugf("select codemode:%d", codeMode)
}

26
blobstore/access/stream/stream_get_test.go
View File

@ -34,7 +34,7 @@ func TestAccessStreamGetBase(t *testing.T) {
{
dataShards.clean()
data := []byte("x")
loc, err := streamer.Put(ctx(), bytes.NewReader(data), int64(len(data)), nil)
loc, err := streamer.Put(ctx(), bytes.NewReader(data), int64(len(data)), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
buff := bytes.NewBuffer(nil)
@ -47,7 +47,7 @@ func TestAccessStreamGetBase(t *testing.T) {
{
dataShards.clean()
data := []byte("x")
loc, err := streamer.Put(ctx(), bytes.NewReader(data), int64(len(data)), nil)
loc, err := streamer.Put(ctx(), bytes.NewReader(data), int64(len(data)), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
buff := bytes.NewBuffer(nil)
@ -83,7 +83,7 @@ func TestAccessStreamGetBase(t *testing.T) {
size := cs.size
data := make([]byte, size)
rand.Read(data)
loc, err := streamer.Put(ctx(), bytes.NewReader(data), int64(size), nil)
loc, err := streamer.Put(ctx(), bytes.NewReader(data), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
buff := bytes.NewBuffer(nil)
@ -117,7 +117,7 @@ func TestAccessStreamGetBroken(t *testing.T) {
rand.Read(data)
// time wait the punished services
time.Sleep(time.Second * time.Duration(punishServiceS))
loc, err := streamer.Put(ctx(), bytes.NewReader(data), int64(size), nil)
loc, err := streamer.Put(ctx(), bytes.NewReader(data), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
cases := []struct {
@ -183,7 +183,7 @@ func TestAccessStreamGetOffset(t *testing.T) {
size := cs.size
data := make([]byte, size)
rand.Read(data)
loc, err := streamer.Put(ctx(), bytes.NewReader(data), size, nil)
loc, err := streamer.Put(ctx(), bytes.NewReader(data), size, nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
buff := bytes.NewBuffer(nil)
@ -210,7 +210,7 @@ func TestAccessStreamGetShardTimeout(t *testing.T) {
size := 1 << 22
buff := make([]byte, size)
rand.Read(buff)
loc, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil)
loc, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
// no delay when blocking one shard, cos MinReadShardsX = 1
@ -259,7 +259,7 @@ func TestAccessStreamGetShardSlow(t *testing.T) {
size := 1 << 20
buff := make([]byte, size)
rand.Read(buff)
loc, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil)
loc, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
vuidController.SetSlowdown(1001, 500*time.Millisecond)
@ -287,7 +287,7 @@ func TestAccessStreamGetShardCrcMismatch(t *testing.T) {
dataShards.clean()
buff := make([]byte, size)
rand.Read(buff)
loc, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil)
loc, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
transfer, err := streamer.Get(ctx(), bytes.NewBuffer(nil), *loc, uint64(size), 0)
@ -311,7 +311,7 @@ func TestAccessStreamGetShardBroken(t *testing.T) {
size := 1 << 22
buff := make([]byte, size)
rand.Read(buff)
loc, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil)
loc, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
// no delay when blocking one shard, cos MinReadShardsX = 1
@ -352,7 +352,7 @@ func TestAccessStreamGetShardOnlyTimeout(t *testing.T) {
size := 1
buff := make([]byte, size)
rand.Read(buff)
loc, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil)
loc, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
// blocking the data shard, force to waiting ReadDataOnlyTimeoutMS
@ -384,7 +384,7 @@ func TestAccessStreamGetLocalIDC(t *testing.T) {
size := 1 << 22
buff := make([]byte, size)
rand.Read(buff)
loc, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil)
loc, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
// no delay when blocking other idc all shards
@ -495,7 +495,7 @@ func TestAccessStreamGetAligned(t *testing.T) {
data := make([]byte, cs.size)
rand.Read(data)
loc, err := streamer.Put(ctx(), bytes.NewReader(data), int64(cs.size), nil)
loc, err := streamer.Put(ctx(), bytes.NewReader(data), int64(cs.size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
// cos put shards asynchronously, should wait all shard written
@ -675,7 +675,7 @@ func BenchmarkAccessStreamGet(b *testing.B) {
for _, cs := range cases {
b.ResetTimer()
b.Run(cs.name, func(b *testing.B) {
loc, err := streamer.Put(ctx, newReader(cs.size), int64(cs.size), nil)
loc, err := streamer.Put(ctx, newReader(cs.size), int64(cs.size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(b, err)
b.ResetTimer()

8
blobstore/access/stream/stream_mock_test.go
View File

@ -424,7 +424,13 @@ func initMockData() {
c := NewMockClusterController(ctr)
c.EXPECT().Region().AnyTimes().Return("test-region")
c.EXPECT().ChooseOne().AnyTimes().Return(clusterInfo, nil)
c.EXPECT().GetServiceController(gomock.Any()).AnyTimes().Return(serviceController, nil)
c.EXPECT().GetServiceController(gomock.Any()).AnyTimes().DoAndReturn(
func(needClusterID proto.ClusterID) (controller.ServiceController, error) {
if needClusterID != clusterID {
return nil, fmt.Errorf("no service controller of %d", needClusterID)
}
return serviceController, nil
})
c.EXPECT().GetVolumeGetter(gomock.Any()).AnyTimes().Return(volumeGetter, nil)
c.EXPECT().ChangeChooseAlg(gomock.Any()).AnyTimes().DoAndReturn(
func(alg controller.AlgChoose) error {

17
blobstore/access/stream/stream_put.go
View File

@ -28,6 +28,7 @@ import (
"github.com/cubefs/cubefs/blobstore/api/access"
"github.com/cubefs/cubefs/blobstore/api/blobnode"
"github.com/cubefs/cubefs/blobstore/common/codemode"
"github.com/cubefs/cubefs/blobstore/common/ec"
errcode "github.com/cubefs/cubefs/blobstore/common/errors"
"github.com/cubefs/cubefs/blobstore/common/proto"
@ -43,6 +44,7 @@ import (
// optional: hasher map to calculate hash.Hash
func (h *Handler) Put(ctx context.Context,
rc io.Reader, size int64, hasherMap access.HasherMap,
assignClusterID proto.ClusterID, codeMode codemode.CodeMode,
) (*proto.Location, error) {
span := trace.SpanFromContextSafe(ctx)
span.Debugf("put request size:%d hashes:b(%b)", size, hasherMap.ToHashAlgorithm())
@ -61,11 +63,20 @@ func (h *Handler) Put(ctx context.Context,
}
// 2.choose cluster and alloc volume from allocator
selectedCodeMode := h.allCodeModes.SelectCodeMode(size)
span.Debugf("select codemode %d", selectedCodeMode)
selectedCodeMode := codeMode
if selectedCodeMode == codemode.CodeModeNone {
selectedCodeMode = h.allCodeModes.SelectCodeMode(size)
} else {
valid := h.allCodeModes.VerifySelectCodeMode(selectedCodeMode)
if !valid {
span.Errorf("specify codemode %d not found in codemode policy", selectedCodeMode)
return nil, errcode.ErrIllegalArguments
}
}
span.Debugf("select codemode %d, specify codemode %d", selectedCodeMode, codeMode)
blobSize := atomic.LoadUint32(&h.MaxBlobSize)
clusterID, blobs, err := h.allocFromAllocatorWithHystrix(ctx, selectedCodeMode, uint64(size), blobSize, 0)
clusterID, blobs, err := h.allocFromAllocatorWithHystrix(ctx, selectedCodeMode, uint64(size), blobSize, assignClusterID)
if err != nil {
span.Error("alloc failed", errors.Detail(err))
return nil, err

66
blobstore/access/stream/stream_put_test.go
View File

@ -28,6 +28,7 @@ import (
"github.com/stretchr/testify/require"
"github.com/cubefs/cubefs/blobstore/api/access"
"github.com/cubefs/cubefs/blobstore/common/codemode"
errcode "github.com/cubefs/cubefs/blobstore/common/errors"
"github.com/cubefs/cubefs/blobstore/common/proto"
)
@ -42,13 +43,13 @@ func TestAccessStreamPutBase(t *testing.T) {
// 0
{
size := 0
_, err := streamer.Put(ctx(), newReader(size), int64(size), nil)
_, err := streamer.Put(ctx(), newReader(size), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.Error(t, err)
}
// 1 byte
{
size := 1
loc, err := streamer.Put(ctx(), newReader(size), int64(size), nil)
loc, err := streamer.Put(ctx(), newReader(size), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
require.Equal(t, 1, len(loc.Slices))
require.Equal(t, uint32(1), loc.Slices[0].Count)
@ -58,7 +59,7 @@ func TestAccessStreamPutBase(t *testing.T) {
// <4M
{
size := 1 << 18
loc, err := streamer.Put(ctx(), newReader(size), int64(size), nil)
loc, err := streamer.Put(ctx(), newReader(size), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
require.Equal(t, 1, len(loc.Slices))
require.Equal(t, uint32(1), loc.Slices[0].Count)
@ -67,7 +68,7 @@ func TestAccessStreamPutBase(t *testing.T) {
// 8M + 1k
{
size := (1 << 23) + 1024
loc, err := streamer.Put(ctx(), newReader(size), int64(size), nil)
loc, err := streamer.Put(ctx(), newReader(size), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
require.Equal(t, 2, len(loc.Slices))
require.Equal(t, uint32(2), loc.Slices[1].Count)
@ -76,7 +77,7 @@ func TestAccessStreamPutBase(t *testing.T) {
// max size + 1
{
size := defaultMaxObjectSize + 1
_, err := streamer.Put(ctx(), nil, int64(size), nil)
_, err := streamer.Put(ctx(), nil, int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.EqualError(t, errcode.ErrAccessExceedSize, err.Error())
}
@ -97,7 +98,7 @@ func TestAccessStreamPutSum(t *testing.T) {
}
hashSumMap := make(access.HashSumMap, len(hasherMap))
_, err := streamer.Put(ctx(), bytes.NewReader(data), int64(len(data)), hasherMap)
_, err := streamer.Put(ctx(), bytes.NewReader(data), int64(len(data)), hasherMap, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
for alg, hasher := range hasherMap {
hashSumMap[alg] = hasher.Sum(nil)
@ -231,7 +232,7 @@ func TestAccessStreamPutShardTimeout(t *testing.T) {
buff := make([]byte, size)
rand.Read(buff)
startTime := time.Now()
loc, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil)
loc, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
// response immediately if had quorum shards
@ -247,7 +248,7 @@ func TestAccessStreamPutShardTimeout(t *testing.T) {
vuidController.Block(1002)
{
startTime := time.Now()
_, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil)
_, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.Error(t, err)
duration := time.Since(startTime)
@ -273,7 +274,7 @@ func TestAccessStreamPutShardSlow(t *testing.T) {
size := 3
buff := make([]byte, size)
rand.Read(buff)
_, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil)
_, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
time.Sleep(time.Second)
}
@ -318,7 +319,7 @@ func TestAccessStreamPutQuorum(t *testing.T) {
vuidController.Break(id)
}
_, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil)
_, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
if cs.hasError {
require.NotNil(t, err)
} else {
@ -332,6 +333,49 @@ func TestAccessStreamPutQuorum(t *testing.T) {
}
}
func TestAccessStreamPutWithClusterIDAndCodeMode(t *testing.T) {
ctx := ctxWithName("TestAccessStreamPutWithClusterIDAndCodeMode")
{
size := 1 << 22
_, err := streamer.Put(ctx(), newReader(size), int64(size), nil, proto.ClusterID(0), codemode.EC16P20L2)
require.Error(t, err)
}
{
size := 1 << 22
_, err := streamer.Put(ctx(), newReader(size), int64(size), nil, proto.ClusterID(2), codemode.CodeModeNone)
require.Error(t, err)
}
{
size := 1 << 22
_, err := streamer.Put(ctx(), newReader(size), int64(size), nil, proto.ClusterID(2), codemode.EC15P12)
require.Error(t, err)
}
{
size := 1 << 20
loc, err := streamer.Put(ctx(), newReader(size), int64(size), nil, proto.ClusterID(0), codemode.EC6P6)
require.NoError(t, err)
require.Equal(t, codemode.EC6P6, loc.CodeMode)
require.Equal(t, 1, len(loc.Slices))
require.Equal(t, uint32(1), loc.Slices[0].Count)
}
{
size := 1 << 22
loc, err := streamer.Put(ctx(), newReader(size), int64(size), nil, proto.ClusterID(1), codemode.EC6P6)
require.NoError(t, err)
require.Equal(t, codemode.EC6P6, loc.CodeMode)
require.Equal(t, 1, len(loc.Slices))
require.Equal(t, uint32(1), loc.Slices[0].Count)
}
{
size := 1 << 22
loc, err := streamer.Put(ctx(), newReader(size), int64(size), nil, proto.ClusterID(1), codemode.CodeModeNone)
require.NoError(t, err)
require.Equal(t, codemode.EC6P6, loc.CodeMode)
require.Equal(t, 1, len(loc.Slices))
require.Equal(t, uint32(1), loc.Slices[0].Count)
}
}
func BenchmarkAccessStreamPut(b *testing.B) {
ctx := ctxWithName("BenchmarkAccessStreamPut")()
vuidController.Unbreak(1005)
@ -357,7 +401,7 @@ func BenchmarkAccessStreamPut(b *testing.B) {
b.ResetTimer()
b.Run(cs.name, func(b *testing.B) {
for ii := 0; ii <= b.N; ii++ {
streamer.Put(ctx, bytes.NewReader(buff[:cs.size]), int64(cs.size), nil)
streamer.Put(ctx, bytes.NewReader(buff[:cs.size]), int64(cs.size), nil, proto.ClusterID(0), codemode.CodeModeNone)
}
})
}

3
blobstore/access/stream/stream_test.go
View File

@ -24,6 +24,7 @@ import (
"github.com/cubefs/cubefs/blobstore/access/controller"
"github.com/cubefs/cubefs/blobstore/common/codemode"
"github.com/cubefs/cubefs/blobstore/common/proto"
)
func newReader(size int) io.Reader {
@ -79,7 +80,7 @@ func TestAccessStreamNew(t *testing.T) {
func TestAccessStreamDelete(t *testing.T) {
ctx := ctxWithName("TestAccessStreamDelete")
size := 1 << 18
loc, err := streamer.Put(ctx(), newReader(size), int64(size), nil)
loc, err := streamer.Put(ctx(), newReader(size), int64(size), nil, proto.ClusterID(0), codemode.CodeModeNone)
require.NoError(t, err)
err = streamer.Delete(ctx(), loc)

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

@ -424,7 +424,9 @@ func (c *client) Put(ctx context.Context, args *PutArgs) (location proto.Locatio
func (c *client) putObject(ctx context.Context, args *PutArgs) (location proto.Location, hashSumMap HashSumMap, err error) {
rpcClient := c.rpcClient.Load().(rpc.Client)
urlStr := fmt.Sprintf("/put?size=%d&hashes=%d", args.Size, args.Hashes)
urlStr := fmt.Sprintf("/put?size=%d&hashes=%d&assign_cluster_id=%d&code_mode=%d",
args.Size, args.Hashes, args.AssignClusterID, args.CodeMode)
req, err := http.NewRequest(http.MethodPut, urlStr, args.Body)
if err != nil {
return
@ -570,7 +572,12 @@ func (c *client) putParts(ctx context.Context, args *PutArgs) (proto.Location, H
// alloc
allocResp := &AllocResp{}
if err := rpcClient.PostWith(ctx, "/alloc", allocResp, AllocArgs{Size: uint64(args.Size)}); err != nil {
allocArgs := &AllocArgs{
Size: uint64(args.Size),
AssignClusterID: args.AssignClusterID,
CodeMode: args.CodeMode,
}
if err := rpcClient.PostWith(ctx, "/alloc", allocResp, allocArgs); err != nil {
return allocResp.Location, nil, err
}
loc = allocResp.Location

7
blobstore/api/access/proto.go
View File

@ -196,11 +196,18 @@ func (h HashSumMap) All() map[string]interface{} {
// PutArgs for service /put
// Hashes means how to calculate check sum,
// HashAlgCRC32 | HashAlgMD5 equal 2 + 4 = 6
// AssignClusterID > 0 means that cluster_id is assigned by the API caller
// AssignClusterID = 0 means that cluster_id is assigned by access cluster controller
// CodeMode > 0 means that codemode is assigned by the API caller
// CodeMode = 0 means that codemode is assigned by code_mode_policies
type PutArgs struct {
Size int64 `json:"size"`
Hashes HashAlgorithm `json:"hashes,omitempty"`
Body io.Reader `json:"-"`
AssignClusterID proto.ClusterID `json:"assign_cluster_id,omitempty"`
CodeMode codemode.CodeMode `json:"code_mode,omitempty"`
// GetBody defines an optional func to return a new copy of Body.
// It is used for client requests when a redirect requires reading
// the body more than once. Use of GetBody still requires setting Body.

2
blobstore/cli/access/access.go
View File

@ -185,6 +185,8 @@ func Register(app *grumble.App) {
f.String("f", "filepath", "", "put file path")
f.Int64("", "size", 0, "put file size, 0 means file size")
f.Uint("", "hashes", 0, "put file hashes")
f.Uint("", "codemode", 0, "put file codemode")
f.Uint("", "clusterid", 0, "put file cluster id")
},
})
accessCommand.AddCommand(&grumble.Command{

12
blobstore/cli/access/put.go
View File

@ -26,6 +26,8 @@ import (
"github.com/cubefs/cubefs/blobstore/cli/common/flags"
"github.com/cubefs/cubefs/blobstore/cli/common/fmt"
"github.com/cubefs/cubefs/blobstore/cli/config"
"github.com/cubefs/cubefs/blobstore/common/codemode"
"github.com/cubefs/cubefs/blobstore/common/proto"
)
func putFile(c *grumble.Context) error {
@ -69,11 +71,15 @@ func putFile(c *grumble.Context) error {
putHashes := access.HashAlgorithm(c.Flags.Uint("hashes"))
fmt.Printf("to upload size:%d hash:b(%b)\n", size, putHashes)
codeMode := c.Flags.Uint("codemode")
clusterID := c.Flags.Uint("clusterid")
location, hashes, err := client.Put(common.CmdContext(), &access.PutArgs{
Size: size,
Hashes: putHashes,
Body: reader,
Size: size,
Hashes: putHashes,
Body: reader,
AssignClusterID: proto.ClusterID(clusterID),
CodeMode: codemode.CodeMode(codeMode),
})
if err != nil {
return err

1
blobstore/common/codemode/codemode.go
View File

@ -26,6 +26,7 @@ type (
// pre-defined mode
const (
CodeModeNone CodeMode = 0
EC15P12 CodeMode = 1
EC6P6 CodeMode = 2
EC16P20L2 CodeMode = 3

9
blobstore/sdk/sdk_client.go
View File

@ -587,7 +587,7 @@ func (s *sdkHandler) doPutObject(ctx context.Context, args *acapi.PutArgs) (prot
}
rc := s.limiter.Reader(ctx, args.Body)
loc, err := s.handler.Put(ctx, rc, args.Size, hasherMap)
loc, err := s.handler.Put(ctx, rc, args.Size, hasherMap, args.AssignClusterID, args.CodeMode)
if err != nil {
span.Error("stream put failed", errors.Detail(err))
err = httpError(err)
@ -779,7 +779,12 @@ func (s *sdkHandler) putParts(ctx context.Context, args *acapi.PutArgs) (proto.L
}()
// alloc
allocResp, err := s.alloc(ctx, &acapi.AllocArgs{Size: uint64(args.Size)})
allocArgs := &acapi.AllocArgs{
Size: uint64(args.Size),
AssignClusterID: args.AssignClusterID,
CodeMode: args.CodeMode,
}
allocResp, err := s.alloc(ctx, allocArgs)
if err != nil {
return proto.Location{}, nil, err
}

29
blobstore/sdk/sdk_client_test.go
View File

@ -256,7 +256,7 @@ func TestSdkHandler_Put(t *testing.T) {
args := &acapi.PutArgs{Size: 2}
// stream put error
hd.handler.(*mocks.MockStreamHandler).EXPECT().Put(gAny, gAny, gAny, gAny).Return(nil, errMock)
hd.handler.(*mocks.MockStreamHandler).EXPECT().Put(gAny, gAny, gAny, gAny, gAny, gAny).Return(nil, errMock)
loc, hash, err = hd.Put(ctx, args)
require.NotNil(t, err)
require.Equal(t, uint64(0), loc.Size_)
@ -265,7 +265,7 @@ func TestSdkHandler_Put(t *testing.T) {
// ok
args.Hashes = 1
mockLoc := proto.Location{Size_: 2}
hd.handler.(*mocks.MockStreamHandler).EXPECT().Put(gAny, gAny, gAny, gAny).Return(&mockLoc, nil)
hd.handler.(*mocks.MockStreamHandler).EXPECT().Put(gAny, gAny, gAny, gAny, gAny, gAny).Return(&mockLoc, nil)
loc, hash, err = hd.Put(ctx, args)
require.NoError(t, err)
require.Equal(t, mockLoc.Size_, loc.Size_)
@ -275,7 +275,7 @@ func TestSdkHandler_Put(t *testing.T) {
args.GetBody = func() (io.ReadCloser, error) {
return nil, nil
}
hd.handler.(*mocks.MockStreamHandler).EXPECT().Put(gAny, gAny, gAny, gAny).Return(nil, errMock).Times(3)
hd.handler.(*mocks.MockStreamHandler).EXPECT().Put(gAny, gAny, gAny, gAny, gAny, gAny).Return(nil, errMock).Times(3)
loc, hash, err = hd.Put(ctx, args)
require.NotNil(t, err)
require.Equal(t, uint64(0), loc.Size_)
@ -287,12 +287,31 @@ func TestSdkHandler_Put(t *testing.T) {
buff := bytes.NewBuffer(data)
return io.NopCloser(buff), nil
}
hd.handler.(*mocks.MockStreamHandler).EXPECT().Put(gAny, gAny, gAny, gAny).Return(nil, errMock)
hd.handler.(*mocks.MockStreamHandler).EXPECT().Put(gAny, gAny, gAny, gAny).Return(&mockLoc, nil)
hd.handler.(*mocks.MockStreamHandler).EXPECT().Put(gAny, gAny, gAny, gAny, gAny, gAny).Return(nil, errMock)
hd.handler.(*mocks.MockStreamHandler).EXPECT().Put(gAny, gAny, gAny, gAny, gAny, gAny).Return(&mockLoc, nil)
loc, hash, err = hd.Put(ctx, args)
require.NoError(t, err)
require.Equal(t, mockLoc.Size_, loc.Size_)
require.Equal(t, int(args.Hashes), len(hash))
hd.handler.(*mocks.MockStreamHandler).EXPECT().Put(gAny, gAny, gAny, gAny, gAny, gAny).Return(nil, errMock)
data = make([]byte, 1)
argsWithClusterIDAndCodeMode := &acapi.PutArgs{
Size: 1,
Hashes: 1,
Body: bytes.NewBuffer(data),
AssignClusterID: 1,
CodeMode: codemode.EC3P3,
}
loc, _, err = hd.Put(ctx, argsWithClusterIDAndCodeMode)
require.NotNil(t, err)
require.Equal(t, proto.Location{}, loc)
mockLoc2 := proto.Location{ClusterID: 1, CodeMode: codemode.EC3P3, Size_: 1}
hd.handler.(*mocks.MockStreamHandler).EXPECT().Put(gAny, gAny, gAny, gAny, gAny, gAny).Return(&mockLoc2, nil)
loc, _, err = hd.Put(ctx, argsWithClusterIDAndCodeMode)
require.NoError(t, err)
require.Equal(t, mockLoc2, loc)
}
func TestSdkHandler_Alloc(t *testing.T) {

8
blobstore/testing/mocks/access_stream.go
View File

@ -172,18 +172,18 @@ func (mr *MockStreamHandlerMockRecorder) ListBlob(arg0, arg1 interface{}) *gomoc
}
// Put mocks base method.
func (m *MockStreamHandler) Put(arg0 context.Context, arg1 io.Reader, arg2 int64, arg3 access.HasherMap) (*proto.Location, error) {
func (m *MockStreamHandler) Put(arg0 context.Context, arg1 io.Reader, arg2 int64, arg3 access.HasherMap, arg4 proto.ClusterID, arg5 codemode.CodeMode) (*proto.Location, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "Put", arg0, arg1, arg2, arg3)
ret := m.ctrl.Call(m, "Put", arg0, arg1, arg2, arg3, arg4, arg5)
ret0, _ := ret[0].(*proto.Location)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// Put indicates an expected call of Put.
func (mr *MockStreamHandlerMockRecorder) Put(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
func (mr *MockStreamHandlerMockRecorder) Put(arg0, arg1, arg2, arg3, arg4, arg5 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Put", reflect.TypeOf((*MockStreamHandler)(nil).Put), arg0, arg1, arg2, arg3)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Put", reflect.TypeOf((*MockStreamHandler)(nil).Put), arg0, arg1, arg2, arg3, arg4, arg5)
}
// PutAt mocks base method.