forked from TrueCloudLab/frostfs-node
135 lines
3.4 KiB
Go
135 lines
3.4 KiB
Go
package blobtree
|
|
|
|
import (
|
|
"context"
|
|
"os"
|
|
"time"
|
|
|
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util/logicerr"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
|
|
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
|
|
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
|
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
|
"go.opentelemetry.io/otel/attribute"
|
|
"go.opentelemetry.io/otel/trace"
|
|
)
|
|
|
|
func (b *BlobTree) Get(ctx context.Context, prm common.GetPrm) (common.GetRes, error) {
|
|
var (
|
|
startedAt = time.Now()
|
|
success = false
|
|
size = 0
|
|
)
|
|
defer func() {
|
|
b.cfg.metrics.Get(time.Since(startedAt), size, success, prm.StorageID != nil)
|
|
}()
|
|
|
|
_, span := tracing.StartSpanFromContext(ctx, "BlobTree.Get",
|
|
trace.WithAttributes(
|
|
attribute.String("path", b.cfg.rootPath),
|
|
attribute.String("address", prm.Address.EncodeToString()),
|
|
attribute.String("storage_id", storageIDToIdxStringSafe(prm.StorageID)),
|
|
attribute.Bool("raw", prm.Raw),
|
|
))
|
|
defer span.End()
|
|
|
|
res, err := b.get(prm)
|
|
success = err == nil
|
|
size = len(res.RawData)
|
|
return res, err
|
|
}
|
|
|
|
func (b *BlobTree) get(prm common.GetPrm) (common.GetRes, error) {
|
|
if idx, ok := tryParseIdxFromStorageID(prm.StorageID); ok {
|
|
return b.getFromIdx(prm.Address, idx)
|
|
}
|
|
return b.findAndGet(prm.Address)
|
|
}
|
|
|
|
func (b *BlobTree) getFromIdx(addr oid.Address, idx uint64) (common.GetRes, error) {
|
|
dir := b.getDirectoryPath(addr)
|
|
path := b.getFilePath(dir, idx)
|
|
|
|
b.fileLock.RLock(path)
|
|
defer b.fileLock.RUnlock(path)
|
|
|
|
records, err := b.readFileContent(path)
|
|
if err != nil {
|
|
return common.GetRes{}, err
|
|
}
|
|
|
|
for _, record := range records {
|
|
if record.Address.Equals(addr) {
|
|
return b.unmarshalGetRes(record)
|
|
}
|
|
}
|
|
|
|
return common.GetRes{}, logicerr.Wrap(new(apistatus.ObjectNotFound))
|
|
}
|
|
|
|
func (b *BlobTree) unmarshalGetRes(record objectData) (common.GetRes, error) {
|
|
data, err := b.compressor.Decompress(record.Data)
|
|
if err != nil {
|
|
return common.GetRes{}, err
|
|
}
|
|
|
|
obj := objectSDK.New()
|
|
if err := obj.Unmarshal(data); err != nil {
|
|
return common.GetRes{}, err
|
|
}
|
|
return common.GetRes{Object: obj, RawData: data}, nil
|
|
}
|
|
|
|
func (b *BlobTree) findAndGet(addr oid.Address) (common.GetRes, error) {
|
|
dir := b.getDirectoryPath(addr)
|
|
entities, err := os.ReadDir(dir)
|
|
if err != nil {
|
|
if os.IsNotExist(err) {
|
|
return common.GetRes{}, logicerr.Wrap(new(apistatus.ObjectNotFound))
|
|
}
|
|
return common.GetRes{}, err
|
|
}
|
|
for _, entity := range entities {
|
|
if entity.IsDir() {
|
|
continue
|
|
}
|
|
if b.isTempFile(entity.Name()) {
|
|
continue
|
|
}
|
|
idx, err := b.parseIdx(entity.Name())
|
|
if err != nil {
|
|
continue
|
|
}
|
|
path := b.getFilePath(dir, idx)
|
|
res, err := b.tryReadObject(path, addr)
|
|
if err != nil {
|
|
return common.GetRes{}, err
|
|
}
|
|
if res.Object != nil {
|
|
return res, nil
|
|
}
|
|
}
|
|
return common.GetRes{}, logicerr.Wrap(new(apistatus.ObjectNotFound))
|
|
}
|
|
|
|
func (b *BlobTree) tryReadObject(path string, addr oid.Address) (common.GetRes, error) {
|
|
b.fileLock.RLock(path)
|
|
defer b.fileLock.RUnlock(path)
|
|
|
|
records, err := b.readFileContent(path)
|
|
if err != nil {
|
|
return common.GetRes{}, err
|
|
}
|
|
|
|
for _, record := range records {
|
|
if record.Address.Equals(addr) {
|
|
res, err := b.unmarshalGetRes(record)
|
|
if err != nil {
|
|
return common.GetRes{}, err
|
|
}
|
|
return res, nil
|
|
}
|
|
}
|
|
return common.GetRes{}, nil
|
|
}
|