frostfs-s3-gw/internal/frostfs/frostfs.go

479 lines
13 KiB
Go
Raw Permalink Normal View History

package frostfs
import (
"context"
"errors"
"fmt"
"io"
"math"
"strconv"
"time"
objectv2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/layer"
errorsFrost "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/internal/frostfs/errors"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/eacl"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/pool"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/session"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user"
"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
)
// FrostFS represents virtual connection to the FrostFS network.
// It is used to provide an interface to dependent packages
// which work with FrostFS.
type FrostFS struct {
pool *pool.Pool
await pool.WaitParams
owner user.ID
}
const (
defaultPollInterval = time.Second // overrides default value from pool
defaultPollTimeout = 120 * time.Second // same as default value from pool
)
// NewFrostFS creates new FrostFS using provided pool.Pool.
func NewFrostFS(p *pool.Pool, key *keys.PrivateKey) *FrostFS {
await := pool.WaitParams{PollInterval: defaultPollInterval, Timeout: defaultPollTimeout}
var owner user.ID
user.IDFromKey(&owner, key.PrivateKey.PublicKey)
return &FrostFS{
pool: p,
await: await,
owner: owner,
}
}
// TimeToEpoch implements frostfs.FrostFS interface method.
func (x *FrostFS) TimeToEpoch(ctx context.Context, now, futureTime time.Time) (uint64, uint64, error) {
dur := futureTime.Sub(now)
if dur < 0 {
return 0, 0, fmt.Errorf("time '%s' must be in the future (after %s)",
futureTime.Format(time.RFC3339), now.Format(time.RFC3339))
}
networkInfo, err := x.pool.NetworkInfo(ctx)
if err != nil {
return 0, 0, handleObjectError("get network info via client", err)
}
durEpoch := networkInfo.EpochDuration()
if durEpoch == 0 {
return 0, 0, errors.New("epoch duration is missing or zero")
}
curr := networkInfo.CurrentEpoch()
msPerEpoch := durEpoch * uint64(networkInfo.MsPerBlock())
epochLifetime := uint64(dur.Milliseconds()) / msPerEpoch
if uint64(dur.Milliseconds())%msPerEpoch != 0 {
epochLifetime++
}
var epoch uint64
if epochLifetime >= math.MaxUint64-curr {
epoch = math.MaxUint64
} else {
epoch = curr + epochLifetime
}
return curr, epoch, nil
}
// Container implements frostfs.FrostFS interface method.
func (x *FrostFS) Container(ctx context.Context, layerPrm layer.PrmContainer) (*container.Container, error) {
prm := pool.PrmContainerGet{
ContainerID: layerPrm.ContainerID,
Session: layerPrm.SessionToken,
}
res, err := x.pool.GetContainer(ctx, prm)
if err != nil {
return nil, handleObjectError("read container via connection pool", err)
}
return &res, nil
}
// CreateContainer implements frostfs.FrostFS interface method.
func (x *FrostFS) CreateContainer(ctx context.Context, prm layer.PrmContainerCreate) (*layer.ContainerCreateResult, error) {
var cnr container.Container
cnr.Init()
cnr.SetPlacementPolicy(prm.Policy)
cnr.SetOwner(prm.Creator)
cnr.SetBasicACL(prm.BasicACL)
creationTime := prm.CreationTime
if creationTime.IsZero() {
creationTime = time.Now()
}
container.SetCreationTime(&cnr, creationTime)
if prm.Name != "" {
var d container.Domain
d.SetName(prm.Name)
d.SetZone(prm.Zone)
container.WriteDomain(&cnr, d)
container.SetName(&cnr, prm.Name)
}
for i := range prm.AdditionalAttributes {
cnr.SetAttribute(prm.AdditionalAttributes[i][0], prm.AdditionalAttributes[i][1])
}
err := pool.SyncContainerWithNetwork(ctx, &cnr, x.pool)
if err != nil {
return nil, handleObjectError("sync container with the network state", err)
}
prmPut := pool.PrmContainerPut{
ClientParams: client.PrmContainerPut{
Container: &cnr,
Session: prm.SessionToken,
},
WaitParams: &x.await,
}
// send request to save the container
idCnr, err := x.pool.PutContainer(ctx, prmPut)
return &layer.ContainerCreateResult{
ContainerID: idCnr,
HomomorphicHashDisabled: container.IsHomomorphicHashingDisabled(cnr),
}, handleObjectError("save container via connection pool", err)
}
// UserContainers implements frostfs.FrostFS interface method.
func (x *FrostFS) UserContainers(ctx context.Context, layerPrm layer.PrmUserContainers) ([]cid.ID, error) {
prm := pool.PrmContainerList{
OwnerID: layerPrm.UserID,
Session: layerPrm.SessionToken,
}
r, err := x.pool.ListContainers(ctx, prm)
return r, handleObjectError("list user containers via connection pool", err)
}
// SetContainerEACL implements frostfs.FrostFS interface method.
func (x *FrostFS) SetContainerEACL(ctx context.Context, table eacl.Table, sessionToken *session.Container) error {
prm := pool.PrmContainerSetEACL{Table: table, Session: sessionToken, WaitParams: &x.await}
err := x.pool.SetEACL(ctx, prm)
return handleObjectError("save eACL via connection pool", err)
}
// ContainerEACL implements frostfs.FrostFS interface method.
func (x *FrostFS) ContainerEACL(ctx context.Context, layerPrm layer.PrmContainerEACL) (*eacl.Table, error) {
prm := pool.PrmContainerEACL{
ContainerID: layerPrm.ContainerID,
Session: layerPrm.SessionToken,
}
res, err := x.pool.GetEACL(ctx, prm)
if err != nil {
return nil, handleObjectError("read eACL via connection pool", err)
}
return &res, nil
}
// DeleteContainer implements frostfs.FrostFS interface method.
func (x *FrostFS) DeleteContainer(ctx context.Context, id cid.ID, token *session.Container) error {
prm := pool.PrmContainerDelete{ContainerID: id, Session: token, WaitParams: &x.await}
err := x.pool.DeleteContainer(ctx, prm)
return handleObjectError("delete container via connection pool", err)
}
// CreateObject implements frostfs.FrostFS interface method.
func (x *FrostFS) CreateObject(ctx context.Context, prm layer.PrmObjectCreate) (oid.ID, error) {
attrNum := len(prm.Attributes) + 1 // + creation time
if prm.Filepath != "" {
attrNum++
}
attrs := make([]object.Attribute, 0, attrNum)
var a *object.Attribute
a = object.NewAttribute()
a.SetKey(object.AttributeTimestamp)
creationTime := prm.CreationTime
if creationTime.IsZero() {
creationTime = time.Now()
}
a.SetValue(strconv.FormatInt(creationTime.Unix(), 10))
attrs = append(attrs, *a)
for i := range prm.Attributes {
a = object.NewAttribute()
a.SetKey(prm.Attributes[i][0])
a.SetValue(prm.Attributes[i][1])
attrs = append(attrs, *a)
}
if prm.Filepath != "" {
a = object.NewAttribute()
a.SetKey(object.AttributeFilePath)
a.SetValue(prm.Filepath)
attrs = append(attrs, *a)
}
obj := object.New()
obj.SetContainerID(prm.Container)
obj.SetOwnerID(x.owner)
obj.SetAttributes(attrs...)
obj.SetPayloadSize(prm.PayloadSize)
if prm.BearerToken == nil && prm.PrivateKey != nil {
var owner user.ID
user.IDFromKey(&owner, prm.PrivateKey.PublicKey)
obj.SetOwnerID(owner)
}
if len(prm.Locks) > 0 {
lock := new(object.Lock)
lock.WriteMembers(prm.Locks)
objectv2.WriteLock(obj.ToV2(), (objectv2.Lock)(*lock))
}
var prmPut pool.PrmObjectPut
prmPut.SetHeader(*obj)
prmPut.SetPayload(prm.Payload)
prmPut.SetCopiesNumberVector(prm.CopiesNumber)
prmPut.SetClientCut(prm.ClientCut)
prmPut.WithoutHomomorphicHash(prm.WithoutHomomorphicHash)
prmPut.SetBufferMaxSize(prm.BufferMaxSize)
if prm.BearerToken != nil {
prmPut.UseBearer(*prm.BearerToken)
} else {
prmPut.UseKey(prm.PrivateKey)
}
idObj, err := x.pool.PutObject(ctx, prmPut)
return idObj, handleObjectError("save object via connection pool", err)
}
// wraps io.ReadCloser and transforms Read errors related to access violation
// to frostfs.ErrAccessDenied.
type payloadReader struct {
io.ReadCloser
}
func (x payloadReader) Read(p []byte) (int, error) {
n, err := x.ReadCloser.Read(p)
if err != nil && errors.Is(err, io.EOF) {
return n, err
}
return n, handleObjectError("read payload", err)
}
// ReadObject implements frostfs.FrostFS interface method.
func (x *FrostFS) ReadObject(ctx context.Context, prm layer.PrmObjectRead) (*layer.ObjectPart, error) {
var addr oid.Address
addr.SetContainer(prm.Container)
addr.SetObject(prm.Object)
var prmGet pool.PrmObjectGet
prmGet.SetAddress(addr)
if prm.BearerToken != nil {
prmGet.UseBearer(*prm.BearerToken)
} else {
prmGet.UseKey(prm.PrivateKey)
}
if prm.WithHeader {
if prm.WithPayload {
res, err := x.pool.GetObject(ctx, prmGet)
if err != nil {
return nil, handleObjectError("init full object reading via connection pool", err)
}
defer res.Payload.Close()
payload, err := io.ReadAll(res.Payload)
if err != nil {
return nil, handleObjectError("read full object payload", err)
}
res.Header.SetPayload(payload)
return &layer.ObjectPart{
Head: &res.Header,
}, nil
}
var prmHead pool.PrmObjectHead
prmHead.SetAddress(addr)
if prm.BearerToken != nil {
prmHead.UseBearer(*prm.BearerToken)
} else {
prmHead.UseKey(prm.PrivateKey)
}
hdr, err := x.pool.HeadObject(ctx, prmHead)
if err != nil {
return nil, handleObjectError("read object header via connection pool", err)
}
return &layer.ObjectPart{
Head: &hdr,
}, nil
} else if prm.PayloadRange[0]+prm.PayloadRange[1] == 0 {
res, err := x.pool.GetObject(ctx, prmGet)
if err != nil {
return nil, handleObjectError("init full payload range reading via connection pool", err)
}
return &layer.ObjectPart{
Payload: res.Payload,
}, nil
}
var prmRange pool.PrmObjectRange
prmRange.SetAddress(addr)
prmRange.SetOffset(prm.PayloadRange[0])
prmRange.SetLength(prm.PayloadRange[1])
if prm.BearerToken != nil {
prmRange.UseBearer(*prm.BearerToken)
} else {
prmRange.UseKey(prm.PrivateKey)
}
res, err := x.pool.ObjectRange(ctx, prmRange)
if err != nil {
return nil, handleObjectError("init payload range reading via connection pool", err)
}
return &layer.ObjectPart{
Payload: payloadReader{&res},
}, nil
}
// DeleteObject implements frostfs.FrostFS interface method.
func (x *FrostFS) DeleteObject(ctx context.Context, prm layer.PrmObjectDelete) error {
var addr oid.Address
addr.SetContainer(prm.Container)
addr.SetObject(prm.Object)
var prmDelete pool.PrmObjectDelete
prmDelete.SetAddress(addr)
if prm.BearerToken != nil {
prmDelete.UseBearer(*prm.BearerToken)
} else {
prmDelete.UseKey(prm.PrivateKey)
}
err := x.pool.DeleteObject(ctx, prmDelete)
return handleObjectError("mark object removal via connection pool", err)
}
// SearchObjects implements frostfs.FrostFS interface method.
func (x *FrostFS) SearchObjects(ctx context.Context, prm layer.PrmObjectSearch) ([]oid.ID, error) {
filters := object.NewSearchFilters()
filters.AddRootFilter()
if prm.ExactAttribute[0] != "" {
filters.AddFilter(prm.ExactAttribute[0], prm.ExactAttribute[1], object.MatchStringEqual)
}
if prm.FilePrefix != "" {
filters.AddFilter(object.AttributeFileName, prm.FilePrefix, object.MatchCommonPrefix)
}
var prmSearch pool.PrmObjectSearch
prmSearch.SetContainerID(prm.Container)
prmSearch.SetFilters(filters)
if prm.BearerToken != nil {
prmSearch.UseBearer(*prm.BearerToken)
} else {
prmSearch.UseKey(prm.PrivateKey)
}
res, err := x.pool.SearchObjects(ctx, prmSearch)
if err != nil {
return nil, handleObjectError("init object search via connection pool", err)
}
defer res.Close()
var buf []oid.ID
err = res.Iterate(func(id oid.ID) bool {
buf = append(buf, id)
return false
})
return buf, handleObjectError("read object list", err)
}
// ResolverFrostFS represents virtual connection to the FrostFS network.
// It implements resolver.FrostFS.
type ResolverFrostFS struct {
pool *pool.Pool
}
// NewResolverFrostFS creates new ResolverFrostFS using provided pool.Pool.
func NewResolverFrostFS(p *pool.Pool) *ResolverFrostFS {
return &ResolverFrostFS{pool: p}
}
// SystemDNS implements resolver.FrostFS interface method.
func (x *ResolverFrostFS) SystemDNS(ctx context.Context) (string, error) {
networkInfo, err := x.pool.NetworkInfo(ctx)
if err != nil {
return "", handleObjectError("read network info via client", err)
}
domain := networkInfo.RawNetworkParameter("SystemDNS")
if domain == nil {
return "", errors.New("system DNS parameter not found or empty")
}
return string(domain), nil
}
func handleObjectError(msg string, err error) error {
if err == nil {
return nil
}
if reason, ok := errorsFrost.IsErrObjectAccessDenied(err); ok {
return fmt.Errorf("%s: %w: %s", msg, layer.ErrAccessDenied, reason)
}
if errorsFrost.IsTimeoutError(err) {
return fmt.Errorf("%s: %w: %s", msg, layer.ErrGatewayTimeout, err.Error())
}
return fmt.Errorf("%s: %w", msg, err)
}
// PoolStatistic is a mediator which implements authmate.FrostFS through pool.Pool.
type PoolStatistic struct {
pool *pool.Pool
}
// NewPoolStatistic creates new PoolStatistic using provided pool.Pool.
func NewPoolStatistic(p *pool.Pool) *PoolStatistic {
return &PoolStatistic{pool: p}
}
// Statistic implements interface method.
func (x *PoolStatistic) Statistic() pool.Statistic {
return x.pool.Statistic()
}