[#1686] blobstor/*: Return ErrReadOnly for modifying operations

This check should occur on the shard level, but because
blobstor components expose `Open(readOnly bool)` interface,
it is reasonable to expect an error here.

Signed-off-by: Evgenii Stratonikov <evgeniy@morphbits.ru>
This commit is contained in:
Evgenii Stratonikov 2022-08-23 16:04:01 +03:00 committed by fyrchik
parent 6c2d3b020f
commit b9a2055e1c
8 changed files with 117 additions and 1 deletions

View file

@ -14,6 +14,10 @@ import (
// If blobocvnicza ID is specified, only this blobovnicza is processed. // If blobocvnicza ID is specified, only this blobovnicza is processed.
// Otherwise, all Blobovniczas are processed descending weight. // Otherwise, all Blobovniczas are processed descending weight.
func (b *Blobovniczas) Delete(prm common.DeletePrm) (res common.DeleteRes, err error) { func (b *Blobovniczas) Delete(prm common.DeletePrm) (res common.DeleteRes, err error) {
if b.readOnly {
return common.DeleteRes{}, common.ErrReadOnly
}
var bPrm blobovnicza.DeletePrm var bPrm blobovnicza.DeletePrm
bPrm.SetAddress(prm.Address) bPrm.SetAddress(prm.Address)

View file

@ -30,3 +30,23 @@ func TestGeneric(t *testing.T) {
blobstortest.TestAll(t, newTree, 1024, maxObjectSize) blobstortest.TestAll(t, newTree, 1024, maxObjectSize)
} }
func TestControl(t *testing.T) {
const maxObjectSize = 2048
defer func() { _ = os.RemoveAll(t.Name()) }()
var n int
newTree := func(t *testing.T) common.Storage {
dir := filepath.Join(t.Name(), strconv.Itoa(n))
return NewBlobovniczaTree(
WithLogger(zaptest.NewLogger(t)),
WithObjectSizeLimit(maxObjectSize),
WithBlobovniczaShallowWidth(2),
WithBlobovniczaShallowDepth(2),
WithRootPath(dir),
WithBlobovniczaSize(1<<20))
}
blobstortest.TestControl(t, newTree, 1024, maxObjectSize)
}

View file

@ -13,6 +13,10 @@ import (
// //
// returns error if could not save object in any blobovnicza. // returns error if could not save object in any blobovnicza.
func (b *Blobovniczas) Put(prm common.PutPrm) (common.PutRes, error) { func (b *Blobovniczas) Put(prm common.PutPrm) (common.PutRes, error) {
if b.readOnly {
return common.PutRes{}, common.ErrReadOnly
}
if !prm.DontCompress { if !prm.DontCompress {
prm.RawData = b.compression.Compress(prm.RawData) prm.RawData = b.compression.Compress(prm.RawData)
} }

View file

@ -0,0 +1,7 @@
package common
import "errors"
// ErrReadOnly MUST be returned for modifying operations when the storage was opened
// in readonly mode.
var ErrReadOnly = errors.New("opened as read-only")

View file

@ -5,7 +5,10 @@ import (
) )
// Open implements common.Storage. // Open implements common.Storage.
func (*FSTree) Open(bool) error { return nil } func (t *FSTree) Open(ro bool) error {
t.readOnly = ro
return nil
}
// Init implements common.Storage. // Init implements common.Storage.
func (t *FSTree) Init() error { func (t *FSTree) Init() error {

View file

@ -25,6 +25,8 @@ type FSTree struct {
*compression.Config *compression.Config
Depth int Depth int
DirNameLen int DirNameLen int
readOnly bool
} }
// Info groups the information about file storage. // Info groups the information about file storage.
@ -182,6 +184,10 @@ func (t *FSTree) treePath(addr oid.Address) string {
// Delete removes the object with the specified address from the storage. // Delete removes the object with the specified address from the storage.
func (t *FSTree) Delete(prm common.DeletePrm) (common.DeleteRes, error) { func (t *FSTree) Delete(prm common.DeletePrm) (common.DeleteRes, error) {
if t.readOnly {
return common.DeleteRes{}, common.ErrReadOnly
}
p, err := t.getPath(prm.Address) p, err := t.getPath(prm.Address)
if err != nil { if err != nil {
if os.IsNotExist(err) { if os.IsNotExist(err) {
@ -218,6 +224,10 @@ func (t *FSTree) getPath(addr oid.Address) (string, error) {
// Put puts an object in the storage. // Put puts an object in the storage.
func (t *FSTree) Put(prm common.PutPrm) (common.PutRes, error) { func (t *FSTree) Put(prm common.PutPrm) (common.PutRes, error) {
if t.readOnly {
return common.PutRes{}, common.ErrReadOnly
}
p := t.treePath(prm.Address) p := t.treePath(prm.Address)
if err := util.MkdirAllX(filepath.Dir(p), t.Permissions); err != nil { if err := util.MkdirAllX(filepath.Dir(p), t.Permissions); err != nil {
@ -231,6 +241,10 @@ func (t *FSTree) Put(prm common.PutPrm) (common.PutRes, error) {
// PutStream puts executes handler on a file opened for write. // PutStream puts executes handler on a file opened for write.
func (t *FSTree) PutStream(addr oid.Address, handler func(*os.File) error) error { func (t *FSTree) PutStream(addr oid.Address, handler func(*os.File) error) error {
if t.readOnly {
return common.ErrReadOnly
}
p := t.treePath(addr) p := t.treePath(addr)
if err := util.MkdirAllX(filepath.Dir(p), t.Permissions); err != nil { if err := util.MkdirAllX(filepath.Dir(p), t.Permissions); err != nil {

View file

@ -24,3 +24,18 @@ func TestGeneric(t *testing.T) {
blobstortest.TestAll(t, newTree, 2048, 16*1024) blobstortest.TestAll(t, newTree, 2048, 16*1024)
} }
func TestControl(t *testing.T) {
defer func() { _ = os.RemoveAll(t.Name()) }()
var n int
newTree := func(t *testing.T) common.Storage {
dir := filepath.Join(t.Name(), strconv.Itoa(n))
return New(
WithPath(dir),
WithDepth(2),
WithDirNameLen(2))
}
blobstortest.TestControl(t, newTree, 2048, 2048)
}

View file

@ -0,0 +1,49 @@
package blobstortest
import (
"math/rand"
"testing"
objectCore "github.com/nspcc-dev/neofs-node/pkg/core/object"
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/common"
"github.com/stretchr/testify/require"
)
// TestControl checks correctness of a read-only mode.
// cons must return a storage which is NOT opened.
func TestControl(t *testing.T, cons Constructor, min, max uint64) {
s := cons(t)
require.NoError(t, s.Open(false))
require.NoError(t, s.Init())
objects := prepare(t, 10, s, min, max)
require.NoError(t, s.Close())
require.NoError(t, s.Open(true))
for i := range objects {
var prm common.GetPrm
prm.Address = objects[i].addr
prm.StorageID = objects[i].storageID
prm.Raw = true
_, err := s.Get(prm)
require.NoError(t, err)
}
t.Run("put fails", func(t *testing.T) {
var prm common.PutPrm
prm.Object = NewObject(min + uint64(rand.Intn(int(max-min+1))))
prm.Address = objectCore.AddressOf(prm.Object)
_, err := s.Put(prm)
require.ErrorIs(t, err, common.ErrReadOnly)
})
t.Run("delete fails", func(t *testing.T) {
var prm common.DeletePrm
prm.Address = objects[0].addr
prm.StorageID = objects[0].storageID
_, err := s.Delete(prm)
require.ErrorIs(t, err, common.ErrReadOnly)
})
}