2023-03-03 09:33:08 +00:00
|
|
|
// Package memstore implements a memory-backed common.Storage for testing purposes.
|
|
|
|
package memstore
|
|
|
|
|
|
|
|
import (
|
2023-03-13 11:37:35 +00:00
|
|
|
"context"
|
2023-03-03 09:33:08 +00:00
|
|
|
"fmt"
|
|
|
|
"sync"
|
|
|
|
|
|
|
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
|
|
|
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util/logicerr"
|
|
|
|
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
|
|
|
|
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
|
|
|
)
|
|
|
|
|
|
|
|
const Type = "memstore"
|
|
|
|
|
|
|
|
type memstoreImpl struct {
|
|
|
|
*cfg
|
|
|
|
mu sync.RWMutex
|
|
|
|
objs map[string][]byte
|
|
|
|
}
|
|
|
|
|
|
|
|
func New(opts ...Option) common.Storage {
|
|
|
|
st := &memstoreImpl{
|
|
|
|
cfg: defaultConfig(),
|
|
|
|
objs: map[string][]byte{},
|
|
|
|
}
|
|
|
|
|
|
|
|
for _, opt := range opts {
|
|
|
|
opt(st.cfg)
|
|
|
|
}
|
|
|
|
|
|
|
|
return st
|
|
|
|
}
|
|
|
|
|
2023-03-13 11:37:35 +00:00
|
|
|
func (s *memstoreImpl) Get(_ context.Context, req common.GetPrm) (common.GetRes, error) {
|
2023-03-03 09:33:08 +00:00
|
|
|
key := req.Address.EncodeToString()
|
|
|
|
|
|
|
|
s.mu.RLock()
|
|
|
|
data, exists := s.objs[key]
|
|
|
|
s.mu.RUnlock()
|
|
|
|
|
|
|
|
if !exists {
|
2023-08-04 11:14:07 +00:00
|
|
|
return common.GetRes{}, logicerr.Wrap(new(apistatus.ObjectNotFound))
|
2023-03-03 09:33:08 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// Decompress the data.
|
|
|
|
var err error
|
|
|
|
if data, err = s.compression.Decompress(data); err != nil {
|
|
|
|
return common.GetRes{}, fmt.Errorf("could not decompress object data: %w", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Unmarshal the SDK object.
|
|
|
|
obj := objectSDK.New()
|
|
|
|
if err := obj.Unmarshal(data); err != nil {
|
|
|
|
return common.GetRes{}, fmt.Errorf("could not unmarshal the object: %w", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
return common.GetRes{Object: obj, RawData: data}, nil
|
|
|
|
}
|
|
|
|
|
2023-03-13 11:37:35 +00:00
|
|
|
func (s *memstoreImpl) GetRange(ctx context.Context, req common.GetRangePrm) (common.GetRangeRes, error) {
|
|
|
|
getResp, err := s.Get(ctx, common.GetPrm{
|
2023-03-03 09:33:08 +00:00
|
|
|
Address: req.Address,
|
|
|
|
StorageID: req.StorageID,
|
|
|
|
})
|
|
|
|
if err != nil {
|
|
|
|
return common.GetRangeRes{}, err
|
|
|
|
}
|
|
|
|
|
|
|
|
payload := getResp.Object.Payload()
|
|
|
|
from := req.Range.GetOffset()
|
|
|
|
to := from + req.Range.GetLength()
|
|
|
|
|
|
|
|
if pLen := uint64(len(payload)); to < from || pLen < from || pLen < to {
|
2023-08-04 11:14:07 +00:00
|
|
|
return common.GetRangeRes{}, logicerr.Wrap(new(apistatus.ObjectOutOfRange))
|
2023-03-03 09:33:08 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
return common.GetRangeRes{
|
|
|
|
Data: payload[from:to],
|
|
|
|
}, nil
|
|
|
|
}
|
|
|
|
|
2023-03-13 11:37:35 +00:00
|
|
|
func (s *memstoreImpl) Exists(_ context.Context, req common.ExistsPrm) (common.ExistsRes, error) {
|
2023-03-03 09:33:08 +00:00
|
|
|
key := req.Address.EncodeToString()
|
|
|
|
|
|
|
|
s.mu.RLock()
|
|
|
|
defer s.mu.RUnlock()
|
|
|
|
|
|
|
|
_, exists := s.objs[key]
|
|
|
|
return common.ExistsRes{Exists: exists}, nil
|
|
|
|
}
|
|
|
|
|
2023-04-12 14:01:29 +00:00
|
|
|
func (s *memstoreImpl) Put(_ context.Context, req common.PutPrm) (common.PutRes, error) {
|
2023-03-03 09:33:08 +00:00
|
|
|
if s.readOnly {
|
|
|
|
return common.PutRes{}, common.ErrReadOnly
|
|
|
|
}
|
|
|
|
if !req.DontCompress {
|
|
|
|
req.RawData = s.compression.Compress(req.RawData)
|
|
|
|
}
|
|
|
|
|
|
|
|
key := req.Address.EncodeToString()
|
|
|
|
|
|
|
|
s.mu.Lock()
|
|
|
|
defer s.mu.Unlock()
|
|
|
|
|
|
|
|
s.objs[key] = req.RawData
|
|
|
|
return common.PutRes{StorageID: []byte(s.rootPath)}, nil
|
|
|
|
}
|
|
|
|
|
2023-04-12 14:01:29 +00:00
|
|
|
func (s *memstoreImpl) Delete(_ context.Context, req common.DeletePrm) (common.DeleteRes, error) {
|
2023-03-03 09:33:08 +00:00
|
|
|
if s.readOnly {
|
|
|
|
return common.DeleteRes{}, common.ErrReadOnly
|
|
|
|
}
|
|
|
|
|
|
|
|
key := req.Address.EncodeToString()
|
|
|
|
|
|
|
|
s.mu.Lock()
|
|
|
|
defer s.mu.Unlock()
|
|
|
|
|
|
|
|
if _, exists := s.objs[key]; exists {
|
|
|
|
delete(s.objs, key)
|
|
|
|
return common.DeleteRes{}, nil
|
|
|
|
}
|
|
|
|
|
2023-08-04 11:14:07 +00:00
|
|
|
return common.DeleteRes{}, logicerr.Wrap(new(apistatus.ObjectNotFound))
|
2023-03-03 09:33:08 +00:00
|
|
|
}
|
|
|
|
|
2023-05-24 11:09:11 +00:00
|
|
|
func (s *memstoreImpl) Iterate(_ context.Context, req common.IteratePrm) (common.IterateRes, error) {
|
2023-03-03 09:33:08 +00:00
|
|
|
s.mu.RLock()
|
|
|
|
defer s.mu.RUnlock()
|
|
|
|
for k, v := range s.objs {
|
|
|
|
elem := common.IterationElement{
|
|
|
|
ObjectData: v,
|
|
|
|
}
|
|
|
|
if err := elem.Address.DecodeString(string(k)); err != nil {
|
|
|
|
if req.IgnoreErrors {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
return common.IterateRes{}, logicerr.Wrap(fmt.Errorf("(%T) decoding address string %q: %v", s, string(k), err))
|
|
|
|
}
|
|
|
|
var err error
|
|
|
|
if elem.ObjectData, err = s.compression.Decompress(elem.ObjectData); err != nil {
|
|
|
|
if req.IgnoreErrors {
|
|
|
|
if req.ErrorHandler != nil {
|
|
|
|
return common.IterateRes{}, req.ErrorHandler(elem.Address, err)
|
|
|
|
}
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
return common.IterateRes{}, logicerr.Wrap(fmt.Errorf("(%T) decompressing data for address %q: %v", s, elem.Address.String(), err))
|
|
|
|
}
|
|
|
|
switch {
|
|
|
|
case req.Handler != nil:
|
|
|
|
if err := req.Handler(elem); err != nil {
|
|
|
|
return common.IterateRes{}, err
|
|
|
|
}
|
|
|
|
default:
|
|
|
|
if !req.IgnoreErrors {
|
|
|
|
return common.IterateRes{}, logicerr.Wrap(fmt.Errorf("(%T) no Handler or LazyHandler set for IteratePrm", s))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return common.IterateRes{}, nil
|
|
|
|
}
|