From 9eb018672ce5178c566085c702b01ce06a107fd1 Mon Sep 17 00:00:00 2001 From: Evgenii Stratonikov Date: Wed, 6 Jul 2022 16:41:35 +0300 Subject: [PATCH] [#1523] blobstor: Unify parameters for `Put` operation Signed-off-by: Evgenii Stratonikov --- .../blobstor/blobovniczatree/blobovnicza.go | 20 +++---- .../blobovniczatree/blobovnicza_test.go | 6 +- .../blobstor/blobstor_test.go | 8 +-- .../blobstor/common/put.go | 19 +++++++ .../blobstor/exists_test.go | 4 +- .../blobstor/fstree/fstree.go | 8 +-- .../blobstor/fstree/fstree_test.go | 4 +- .../blobstor/iterate_test.go | 3 +- pkg/local_object_storage/blobstor/put.go | 57 +++++++------------ .../shard/control_test.go | 4 +- pkg/local_object_storage/shard/put.go | 10 ++-- pkg/local_object_storage/writecache/flush.go | 14 +++-- .../writecache/persist.go | 7 ++- 13 files changed, 91 insertions(+), 73 deletions(-) create mode 100644 pkg/local_object_storage/blobstor/common/put.go diff --git a/pkg/local_object_storage/blobstor/blobovniczatree/blobovnicza.go b/pkg/local_object_storage/blobstor/blobovniczatree/blobovnicza.go index c29b78268..5f8ad0b94 100644 --- a/pkg/local_object_storage/blobstor/blobovniczatree/blobovnicza.go +++ b/pkg/local_object_storage/blobstor/blobovniczatree/blobovnicza.go @@ -142,10 +142,10 @@ func indexSlice(number uint64) []uint64 { // save object in the maximum weight blobobnicza. // // returns error if could not save object in any blobovnicza. -func (b *Blobovniczas) Put(addr oid.Address, data []byte) (*blobovnicza.ID, error) { - var prm blobovnicza.PutPrm - prm.SetAddress(addr) - prm.SetMarshaledObject(data) +func (b *Blobovniczas) Put(prm common.PutPrm) (common.PutRes, error) { + var putPrm blobovnicza.PutPrm + putPrm.SetAddress(prm.Address) + putPrm.SetMarshaledObject(prm.RawData) var ( fn func(string) (bool, error) @@ -162,7 +162,7 @@ func (b *Blobovniczas) Put(addr oid.Address, data []byte) (*blobovnicza.ID, erro return false, nil } - if _, err := active.blz.Put(prm); err != nil { + if _, err := active.blz.Put(putPrm); err != nil { // check if blobovnicza is full if errors.Is(err, blobovnicza.ErrFull) { b.log.Debug("blobovnicza overflowed", @@ -193,18 +193,18 @@ func (b *Blobovniczas) Put(addr oid.Address, data []byte) (*blobovnicza.ID, erro id = blobovnicza.NewIDFromBytes([]byte(p)) - storagelog.Write(b.log, storagelog.AddressField(addr), storagelog.OpField("Blobovniczas PUT")) + storagelog.Write(b.log, storagelog.AddressField(prm.Address), storagelog.OpField("Blobovniczas PUT")) return true, nil } - if err := b.iterateDeepest(addr, fn); err != nil { - return nil, err + if err := b.iterateDeepest(prm.Address, fn); err != nil { + return common.PutRes{}, err } else if id == nil { - return nil, errPutFailed + return common.PutRes{}, errPutFailed } - return id, nil + return common.PutRes{BlobovniczaID: id}, nil } // Get reads object from blobovnicza tree. diff --git a/pkg/local_object_storage/blobstor/blobovniczatree/blobovnicza_test.go b/pkg/local_object_storage/blobstor/blobovniczatree/blobovnicza_test.go index 8fd77b93f..1006a64f4 100644 --- a/pkg/local_object_storage/blobstor/blobovniczatree/blobovnicza_test.go +++ b/pkg/local_object_storage/blobstor/blobovniczatree/blobovnicza_test.go @@ -73,12 +73,12 @@ func TestBlobovniczas(t *testing.T) { require.NoError(t, err) // save object in blobovnicza - id, err := b.Put(addr, d) + pRes, err := b.Put(common.PutPrm{Address: addr, RawData: d}) require.NoError(t, err, i) // get w/ blobovnicza ID var prm common.GetPrm - prm.BlobovniczaID = id + prm.BlobovniczaID = pRes.BlobovniczaID prm.Address = addr res, err := b.Get(prm) @@ -94,7 +94,7 @@ func TestBlobovniczas(t *testing.T) { // get range w/ blobovnicza ID var rngPrm common.GetRangePrm - rngPrm.BlobovniczaID = id + rngPrm.BlobovniczaID = pRes.BlobovniczaID rngPrm.Address = addr payload := obj.Payload() diff --git a/pkg/local_object_storage/blobstor/blobstor_test.go b/pkg/local_object_storage/blobstor/blobstor_test.go index 5bb69316e..b8700f202 100644 --- a/pkg/local_object_storage/blobstor/blobstor_test.go +++ b/pkg/local_object_storage/blobstor/blobstor_test.go @@ -48,13 +48,13 @@ func TestCompression(t *testing.T) { } testPut := func(t *testing.T, b *BlobStor, i int) { - var prm PutPrm - prm.SetObject(smallObj[i]) + var prm common.PutPrm + prm.Object = smallObj[i] _, err = b.Put(prm) require.NoError(t, err) - prm = PutPrm{} - prm.SetObject(bigObj[i]) + prm = common.PutPrm{} + prm.Object = bigObj[i] _, err = b.Put(prm) require.NoError(t, err) } diff --git a/pkg/local_object_storage/blobstor/common/put.go b/pkg/local_object_storage/blobstor/common/put.go new file mode 100644 index 000000000..223f8b5e0 --- /dev/null +++ b/pkg/local_object_storage/blobstor/common/put.go @@ -0,0 +1,19 @@ +package common + +import ( + "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobovnicza" + objectSDK "github.com/nspcc-dev/neofs-sdk-go/object" + oid "github.com/nspcc-dev/neofs-sdk-go/object/id" +) + +// PutPrm groups the parameters of Put operation. +type PutPrm struct { + Address oid.Address + Object *objectSDK.Object + RawData []byte +} + +// PutRes groups the resulting values of Put operation. +type PutRes struct { + BlobovniczaID *blobovnicza.ID +} diff --git a/pkg/local_object_storage/blobstor/exists_test.go b/pkg/local_object_storage/blobstor/exists_test.go index 59548423d..cb62e08ae 100644 --- a/pkg/local_object_storage/blobstor/exists_test.go +++ b/pkg/local_object_storage/blobstor/exists_test.go @@ -32,8 +32,8 @@ func TestExists(t *testing.T) { } for i := range objects { - var prm PutPrm - prm.SetObject(objects[i]) + var prm common.PutPrm + prm.Object = objects[i] _, err = b.Put(prm) require.NoError(t, err) } diff --git a/pkg/local_object_storage/blobstor/fstree/fstree.go b/pkg/local_object_storage/blobstor/fstree/fstree.go index da753a62e..365b3e04c 100644 --- a/pkg/local_object_storage/blobstor/fstree/fstree.go +++ b/pkg/local_object_storage/blobstor/fstree/fstree.go @@ -206,14 +206,14 @@ func (t *FSTree) Exists(addr oid.Address) (string, error) { } // Put puts an object in the storage. -func (t *FSTree) Put(addr oid.Address, data []byte) error { - p := t.treePath(addr) +func (t *FSTree) Put(prm common.PutPrm) (common.PutRes, error) { + p := t.treePath(prm.Address) if err := util.MkdirAllX(filepath.Dir(p), t.Permissions); err != nil { - return err + return common.PutRes{}, err } - return os.WriteFile(p, data, t.Permissions) + return common.PutRes{}, os.WriteFile(p, prm.RawData, t.Permissions) } // PutStream puts executes handler on a file opened for write. diff --git a/pkg/local_object_storage/blobstor/fstree/fstree_test.go b/pkg/local_object_storage/blobstor/fstree/fstree_test.go index 273f9c28d..76978b77a 100644 --- a/pkg/local_object_storage/blobstor/fstree/fstree_test.go +++ b/pkg/local_object_storage/blobstor/fstree/fstree_test.go @@ -47,7 +47,9 @@ func TestFSTree(t *testing.T) { data := make([]byte, 10) _, _ = rand.Read(data[:]) - require.NoError(t, fs.Put(a, data)) + + _, err := fs.Put(common.PutPrm{Address: a, RawData: data}) + require.NoError(t, err) store[a.EncodeToString()] = data } diff --git a/pkg/local_object_storage/blobstor/iterate_test.go b/pkg/local_object_storage/blobstor/iterate_test.go index be920441d..06cc6684f 100644 --- a/pkg/local_object_storage/blobstor/iterate_test.go +++ b/pkg/local_object_storage/blobstor/iterate_test.go @@ -6,6 +6,7 @@ import ( "testing" "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobovnicza" + "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/common" oid "github.com/nspcc-dev/neofs-sdk-go/object/id" oidtest "github.com/nspcc-dev/neofs-sdk-go/object/id/test" "github.com/stretchr/testify/require" @@ -66,7 +67,7 @@ func TestIterateObjects(t *testing.T) { } for _, v := range mObjs { - _, err := blobStor.PutRaw(v.addr, v.data, true) + _, err := blobStor.PutRaw(common.PutPrm{Address: v.addr, RawData: v.data}, true) require.NoError(t, err) } diff --git a/pkg/local_object_storage/blobstor/put.go b/pkg/local_object_storage/blobstor/put.go index af05b69c8..1869b04a6 100644 --- a/pkg/local_object_storage/blobstor/put.go +++ b/pkg/local_object_storage/blobstor/put.go @@ -6,21 +6,11 @@ import ( "github.com/klauspost/compress/zstd" "github.com/nspcc-dev/neofs-node/pkg/core/object" + "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/common" storagelog "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/internal/log" objectSDK "github.com/nspcc-dev/neofs-sdk-go/object" - oid "github.com/nspcc-dev/neofs-sdk-go/object/id" ) -// PutPrm groups the parameters of Put operation. -type PutPrm struct { - rwObject -} - -// PutRes groups the resulting values of Put operation. -type PutRes struct { - roBlobovniczaID -} - // Put saves the object in BLOB storage. // // If object is "big", BlobStor saves the object in shallow dir. @@ -29,14 +19,18 @@ type PutRes struct { // // Returns any error encountered that // did not allow to completely save the object. -func (b *BlobStor) Put(prm PutPrm) (PutRes, error) { - // marshal object - data, err := prm.obj.Marshal() - if err != nil { - return PutRes{}, fmt.Errorf("could not marshal the object: %w", err) +func (b *BlobStor) Put(prm common.PutPrm) (common.PutRes, error) { + prm.Address = object.AddressOf(prm.Object) + if prm.RawData == nil { + // marshal object + data, err := prm.Object.Marshal() + if err != nil { + return common.PutRes{}, fmt.Errorf("could not marshal the object: %w", err) + } + prm.RawData = data } - return b.PutRaw(object.AddressOf(prm.obj), data, b.NeedsCompression(prm.obj)) + return b.PutRaw(prm, b.NeedsCompression(prm.Object)) } // NeedsCompression returns true if the object should be compressed. @@ -48,46 +42,37 @@ func (b *BlobStor) NeedsCompression(obj *objectSDK.Object) bool { } // PutRaw saves an already marshaled object in BLOB storage. -func (b *BlobStor) PutRaw(addr oid.Address, data []byte, compress bool) (PutRes, error) { - big := b.isBig(data) +func (b *BlobStor) PutRaw(prm common.PutPrm, compress bool) (common.PutRes, error) { + big := b.isBig(prm.RawData) if big { var err error if compress { - err = b.fsTree.PutStream(addr, func(f *os.File) error { + err = b.fsTree.PutStream(prm.Address, func(f *os.File) error { enc, _ := zstd.NewWriter(f) // nil error if no options are provided - if _, err := enc.Write(data); err != nil { + if _, err := enc.Write(prm.RawData); err != nil { return err } return enc.Close() }) } else { - err = b.fsTree.Put(addr, data) + _, err = b.fsTree.Put(prm) } if err != nil { - return PutRes{}, err + return common.PutRes{}, err } - storagelog.Write(b.log, storagelog.AddressField(addr), storagelog.OpField("fstree PUT")) + storagelog.Write(b.log, storagelog.AddressField(prm.Address), storagelog.OpField("fstree PUT")) - return PutRes{}, nil + return common.PutRes{}, nil } if compress { - data = b.CConfig.Compress(data) + prm.RawData = b.CConfig.Compress(prm.RawData) } // save object in blobovnicza - res, err := b.blobovniczas.Put(addr, data) - if err != nil { - return PutRes{}, err - } - - return PutRes{ - roBlobovniczaID: roBlobovniczaID{ - blobovniczaID: res, - }, - }, nil + return b.blobovniczas.Put(prm) } // checks if object is "big". diff --git a/pkg/local_object_storage/shard/control_test.go b/pkg/local_object_storage/shard/control_test.go index b5b0d24b1..21bb4d91c 100644 --- a/pkg/local_object_storage/shard/control_test.go +++ b/pkg/local_object_storage/shard/control_test.go @@ -8,6 +8,7 @@ import ( "github.com/nspcc-dev/neofs-node/pkg/core/object" "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor" + "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/common" "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/fstree" meta "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/metabase" "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/pilorama" @@ -109,7 +110,8 @@ func TestRefillMetabaseCorrupted(t *testing.T) { Depth: 1, Info: sh.blobStor.DumpInfo(), } - require.NoError(t, fs.Put(addr, []byte("not an object"))) + _, err = fs.Put(common.PutPrm{Address: addr, RawData: []byte("not an object")}) + require.NoError(t, err) sh = New( WithBlobStorOptions(blobOpts...), diff --git a/pkg/local_object_storage/shard/put.go b/pkg/local_object_storage/shard/put.go index 87fcfcd25..c8834ae01 100644 --- a/pkg/local_object_storage/shard/put.go +++ b/pkg/local_object_storage/shard/put.go @@ -3,7 +3,7 @@ package shard import ( "fmt" - "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor" + "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/common" meta "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/metabase" "github.com/nspcc-dev/neofs-sdk-go/object" "go.uber.org/zap" @@ -34,8 +34,8 @@ func (s *Shard) Put(prm PutPrm) (PutRes, error) { return PutRes{}, ErrReadOnlyMode } - var putPrm blobstor.PutPrm // form Put parameters - putPrm.SetObject(prm.obj) + var putPrm common.PutPrm // form Put parameters + putPrm.Object = prm.obj // exist check are not performed there, these checks should be executed // ahead of `Put` by storage engine @@ -51,7 +51,7 @@ func (s *Shard) Put(prm PutPrm) (PutRes, error) { var ( err error - res blobstor.PutRes + res common.PutRes ) if res, err = s.blobStor.Put(putPrm); err != nil { @@ -61,7 +61,7 @@ func (s *Shard) Put(prm PutPrm) (PutRes, error) { if !m.NoMetabase() { var pPrm meta.PutPrm pPrm.SetObject(prm.obj) - pPrm.SetBlobovniczaID(res.BlobovniczaID()) + pPrm.SetBlobovniczaID(res.BlobovniczaID) if _, err := s.metaBase.Put(pPrm); err != nil { // may we need to handle this case in a special way // since the object has been successfully written to BlobStor diff --git a/pkg/local_object_storage/writecache/flush.go b/pkg/local_object_storage/writecache/flush.go index 770ff5709..a626bf715 100644 --- a/pkg/local_object_storage/writecache/flush.go +++ b/pkg/local_object_storage/writecache/flush.go @@ -6,7 +6,7 @@ import ( "github.com/mr-tron/base58" "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobovnicza" - "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor" + "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/common" "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/fstree" meta "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/metabase" "github.com/nspcc-dev/neofs-sdk-go/object" @@ -154,7 +154,11 @@ func (c *cache) flushBigObjects() { _, compress := c.compressFlags[sAddr] c.mtx.Unlock() - if _, err := c.blobstor.PutRaw(addr, data, compress); err != nil { + var prm common.PutPrm + prm.Address = addr + prm.RawData = data + + if _, err := c.blobstor.PutRaw(common.PutPrm{Address: addr, RawData: data}, compress); err != nil { c.log.Error("cant flush object to blobstor", zap.Error(err)) return nil } @@ -227,15 +231,15 @@ func (c *cache) writeObject(obj *object.Object, metaOnly bool) error { var id *blobovnicza.ID if !metaOnly { - var prm blobstor.PutPrm - prm.SetObject(obj) + var prm common.PutPrm + prm.Object = obj res, err := c.blobstor.Put(prm) if err != nil { return err } - id = res.BlobovniczaID() + id = res.BlobovniczaID } var pPrm meta.PutPrm diff --git a/pkg/local_object_storage/writecache/persist.go b/pkg/local_object_storage/writecache/persist.go index deaadbb3e..f0692d289 100644 --- a/pkg/local_object_storage/writecache/persist.go +++ b/pkg/local_object_storage/writecache/persist.go @@ -5,6 +5,7 @@ import ( "time" "github.com/nspcc-dev/neofs-node/pkg/core/object" + "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/common" storagelog "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/internal/log" "go.etcd.io/bbolt" "go.uber.org/zap" @@ -113,7 +114,11 @@ func (c *cache) persistBigObject(objInfo objectInfo) { cacheSz := c.estimateCacheSize() metaIndex := 0 if c.incSizeFS(cacheSz) <= c.maxCacheSize { - err := c.fsTree.Put(object.AddressOf(objInfo.obj), objInfo.data) + var prm common.PutPrm + prm.Address = object.AddressOf(objInfo.obj) + prm.RawData = objInfo.data + + _, err := c.fsTree.Put(prm) if err == nil { metaIndex = 1 if c.blobstor.NeedsCompression(objInfo.obj) {