forked from TrueCloudLab/frostfs-sdk-go
2707 lines
72 KiB
Go
2707 lines
72 KiB
Go
package pool
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/ecdsa"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"math"
|
|
"math/rand"
|
|
"sort"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/accounting"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer"
|
|
sdkClient "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
|
|
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
|
|
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
|
|
frostfsecdsa "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/crypto/ecdsa"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/eacl"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
|
|
"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/object/relations"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/session"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user"
|
|
"github.com/google/uuid"
|
|
"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
|
|
"go.uber.org/zap"
|
|
"go.uber.org/zap/zapcore"
|
|
"google.golang.org/grpc"
|
|
)
|
|
|
|
// client represents virtual connection to the single FrostFS network endpoint from which Pool is formed.
|
|
// This interface is expected to have exactly one production implementation - clientWrapper.
|
|
// Others are expected to be for test purposes only.
|
|
type client interface {
|
|
// see clientWrapper.balanceGet.
|
|
balanceGet(context.Context, PrmBalanceGet) (accounting.Decimal, error)
|
|
// see clientWrapper.containerPut.
|
|
containerPut(context.Context, PrmContainerPut) (cid.ID, error)
|
|
// see clientWrapper.containerGet.
|
|
containerGet(context.Context, PrmContainerGet) (container.Container, error)
|
|
// see clientWrapper.containerList.
|
|
containerList(context.Context, PrmContainerList) ([]cid.ID, error)
|
|
// see clientWrapper.containerDelete.
|
|
containerDelete(context.Context, PrmContainerDelete) error
|
|
// see clientWrapper.containerEACL.
|
|
containerEACL(context.Context, PrmContainerEACL) (eacl.Table, error)
|
|
// see clientWrapper.containerSetEACL.
|
|
containerSetEACL(context.Context, PrmContainerSetEACL) error
|
|
// see clientWrapper.endpointInfo.
|
|
endpointInfo(context.Context, prmEndpointInfo) (netmap.NodeInfo, error)
|
|
// see clientWrapper.networkInfo.
|
|
networkInfo(context.Context, prmNetworkInfo) (netmap.NetworkInfo, error)
|
|
// see clientWrapper.objectPut.
|
|
objectPut(context.Context, PrmObjectPut) (oid.ID, error)
|
|
// see clientWrapper.objectDelete.
|
|
objectDelete(context.Context, PrmObjectDelete) error
|
|
// see clientWrapper.objectGet.
|
|
objectGet(context.Context, PrmObjectGet) (ResGetObject, error)
|
|
// see clientWrapper.objectHead.
|
|
objectHead(context.Context, PrmObjectHead) (object.Object, error)
|
|
// see clientWrapper.objectRange.
|
|
objectRange(context.Context, PrmObjectRange) (ResObjectRange, error)
|
|
// see clientWrapper.objectSearch.
|
|
objectSearch(context.Context, PrmObjectSearch) (ResObjectSearch, error)
|
|
// see clientWrapper.sessionCreate.
|
|
sessionCreate(context.Context, prmCreateSession) (resCreateSession, error)
|
|
|
|
clientStatus
|
|
|
|
// see clientWrapper.dial.
|
|
dial(ctx context.Context) error
|
|
// see clientWrapper.restartIfUnhealthy.
|
|
restartIfUnhealthy(ctx context.Context) (bool, bool)
|
|
}
|
|
|
|
// clientStatus provide access to some metrics for connection.
|
|
type clientStatus interface {
|
|
// isHealthy checks if the connection can handle requests.
|
|
isHealthy() bool
|
|
// setUnhealthy marks client as unhealthy.
|
|
setUnhealthy()
|
|
// address return address of endpoint.
|
|
address() string
|
|
// currentErrorRate returns current errors rate.
|
|
// After specific threshold connection is considered as unhealthy.
|
|
// Pool.startRebalance routine can make this connection healthy again.
|
|
currentErrorRate() uint32
|
|
// overallErrorRate returns the number of all happened errors.
|
|
overallErrorRate() uint64
|
|
// methodsStatus returns statistic for all used methods.
|
|
methodsStatus() []statusSnapshot
|
|
}
|
|
|
|
// errPoolClientUnhealthy is an error to indicate that client in pool is unhealthy.
|
|
var errPoolClientUnhealthy = errors.New("pool client unhealthy")
|
|
|
|
// clientStatusMonitor count error rate and other statistics for connection.
|
|
type clientStatusMonitor struct {
|
|
logger *zap.Logger
|
|
addr string
|
|
healthy *atomic.Bool
|
|
errorThreshold uint32
|
|
|
|
mu sync.RWMutex // protect counters
|
|
currentErrorCount uint32
|
|
overallErrorCount uint64
|
|
methods []*methodStatus
|
|
}
|
|
|
|
// methodStatus provide statistic for specific method.
|
|
type methodStatus struct {
|
|
name string
|
|
mu sync.RWMutex // protect counters
|
|
statusSnapshot
|
|
}
|
|
|
|
// statusSnapshot is statistic for specific method.
|
|
type statusSnapshot struct {
|
|
allTime uint64
|
|
allRequests uint64
|
|
}
|
|
|
|
// MethodIndex index of method in list of statuses in clientStatusMonitor.
|
|
type MethodIndex int
|
|
|
|
const (
|
|
methodBalanceGet MethodIndex = iota
|
|
methodContainerPut
|
|
methodContainerGet
|
|
methodContainerList
|
|
methodContainerDelete
|
|
methodContainerEACL
|
|
methodContainerSetEACL
|
|
methodEndpointInfo
|
|
methodNetworkInfo
|
|
methodObjectPut
|
|
methodObjectDelete
|
|
methodObjectGet
|
|
methodObjectHead
|
|
methodObjectRange
|
|
methodSessionCreate
|
|
methodLast
|
|
)
|
|
|
|
// String implements fmt.Stringer.
|
|
func (m MethodIndex) String() string {
|
|
switch m {
|
|
case methodBalanceGet:
|
|
return "balanceGet"
|
|
case methodContainerPut:
|
|
return "containerPut"
|
|
case methodContainerGet:
|
|
return "containerGet"
|
|
case methodContainerList:
|
|
return "containerList"
|
|
case methodContainerDelete:
|
|
return "containerDelete"
|
|
case methodContainerEACL:
|
|
return "containerEACL"
|
|
case methodContainerSetEACL:
|
|
return "containerSetEACL"
|
|
case methodEndpointInfo:
|
|
return "endpointInfo"
|
|
case methodNetworkInfo:
|
|
return "networkInfo"
|
|
case methodObjectPut:
|
|
return "objectPut"
|
|
case methodObjectDelete:
|
|
return "objectDelete"
|
|
case methodObjectGet:
|
|
return "objectGet"
|
|
case methodObjectHead:
|
|
return "objectHead"
|
|
case methodObjectRange:
|
|
return "objectRange"
|
|
case methodSessionCreate:
|
|
return "sessionCreate"
|
|
case methodLast:
|
|
return "it's a system name rather than a method"
|
|
default:
|
|
return "unknown"
|
|
}
|
|
}
|
|
|
|
func newClientStatusMonitor(logger *zap.Logger, addr string, errorThreshold uint32) clientStatusMonitor {
|
|
methods := make([]*methodStatus, methodLast)
|
|
for i := methodBalanceGet; i < methodLast; i++ {
|
|
methods[i] = &methodStatus{name: i.String()}
|
|
}
|
|
|
|
healthy := new(atomic.Bool)
|
|
healthy.Store(true)
|
|
|
|
return clientStatusMonitor{
|
|
logger: logger,
|
|
addr: addr,
|
|
healthy: healthy,
|
|
errorThreshold: errorThreshold,
|
|
methods: methods,
|
|
}
|
|
}
|
|
|
|
func (m *methodStatus) snapshot() statusSnapshot {
|
|
m.mu.RLock()
|
|
defer m.mu.RUnlock()
|
|
return m.statusSnapshot
|
|
}
|
|
|
|
func (m *methodStatus) incRequests(elapsed time.Duration) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.allTime += uint64(elapsed)
|
|
m.allRequests++
|
|
}
|
|
|
|
// clientWrapper is used by default, alternative implementations are intended for testing purposes only.
|
|
type clientWrapper struct {
|
|
clientMutex sync.RWMutex
|
|
client *sdkClient.Client
|
|
prm wrapperPrm
|
|
|
|
clientStatusMonitor
|
|
}
|
|
|
|
// wrapperPrm is params to create clientWrapper.
|
|
type wrapperPrm struct {
|
|
logger *zap.Logger
|
|
address string
|
|
key ecdsa.PrivateKey
|
|
dialTimeout time.Duration
|
|
streamTimeout time.Duration
|
|
errorThreshold uint32
|
|
responseInfoCallback func(sdkClient.ResponseMetaInfo) error
|
|
poolRequestInfoCallback func(RequestInfo)
|
|
dialOptions []grpc.DialOption
|
|
}
|
|
|
|
// setAddress sets endpoint to connect in FrostFS network.
|
|
func (x *wrapperPrm) setAddress(address string) {
|
|
x.address = address
|
|
}
|
|
|
|
// setKey sets sdkClient.Client private key to be used for the protocol communication by default.
|
|
func (x *wrapperPrm) setKey(key ecdsa.PrivateKey) {
|
|
x.key = key
|
|
}
|
|
|
|
// setDialTimeout sets the timeout for connection to be established.
|
|
func (x *wrapperPrm) setDialTimeout(timeout time.Duration) {
|
|
x.dialTimeout = timeout
|
|
}
|
|
|
|
// setStreamTimeout sets the timeout for individual operations in streaming RPC.
|
|
func (x *wrapperPrm) setStreamTimeout(timeout time.Duration) {
|
|
x.streamTimeout = timeout
|
|
}
|
|
|
|
// setErrorThreshold sets threshold after reaching which connection is considered unhealthy
|
|
// until Pool.startRebalance routing updates its status.
|
|
func (x *wrapperPrm) setErrorThreshold(threshold uint32) {
|
|
x.errorThreshold = threshold
|
|
}
|
|
|
|
// setPoolRequestCallback sets callback that will be invoked after every pool response.
|
|
func (x *wrapperPrm) setPoolRequestCallback(f func(RequestInfo)) {
|
|
x.poolRequestInfoCallback = f
|
|
}
|
|
|
|
// setResponseInfoCallback sets callback that will be invoked after every response.
|
|
func (x *wrapperPrm) setResponseInfoCallback(f func(sdkClient.ResponseMetaInfo) error) {
|
|
x.responseInfoCallback = f
|
|
}
|
|
|
|
// setGRPCDialOptions sets the gRPC dial options for new gRPC client connection.
|
|
func (x *wrapperPrm) setGRPCDialOptions(opts []grpc.DialOption) {
|
|
x.dialOptions = opts
|
|
}
|
|
|
|
// newWrapper creates a clientWrapper that implements the client interface.
|
|
func newWrapper(prm wrapperPrm) *clientWrapper {
|
|
var cl sdkClient.Client
|
|
var prmInit sdkClient.PrmInit
|
|
prmInit.SetDefaultPrivateKey(prm.key)
|
|
prmInit.SetResponseInfoCallback(prm.responseInfoCallback)
|
|
|
|
cl.Init(prmInit)
|
|
|
|
res := &clientWrapper{
|
|
client: &cl,
|
|
clientStatusMonitor: newClientStatusMonitor(prm.logger, prm.address, prm.errorThreshold),
|
|
prm: prm,
|
|
}
|
|
|
|
return res
|
|
}
|
|
|
|
// dial establishes a connection to the server from the FrostFS network.
|
|
// Returns an error describing failure reason. If failed, the client
|
|
// SHOULD NOT be used.
|
|
func (c *clientWrapper) dial(ctx context.Context) error {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
var prmDial sdkClient.PrmDial
|
|
prmDial.SetServerURI(c.prm.address)
|
|
prmDial.SetTimeout(c.prm.dialTimeout)
|
|
prmDial.SetStreamTimeout(c.prm.streamTimeout)
|
|
prmDial.SetGRPCDialOptions(c.prm.dialOptions...)
|
|
|
|
if err = cl.Dial(ctx, prmDial); err != nil {
|
|
c.setUnhealthy()
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// restartIfUnhealthy checks healthy status of client and recreate it if status is unhealthy.
|
|
// Return current healthy status and indicating if status was changed by this function call.
|
|
func (c *clientWrapper) restartIfUnhealthy(ctx context.Context) (healthy, changed bool) {
|
|
var wasHealthy bool
|
|
if _, err := c.endpointInfo(ctx, prmEndpointInfo{}); err == nil {
|
|
return true, false
|
|
} else if !errors.Is(err, errPoolClientUnhealthy) {
|
|
wasHealthy = true
|
|
}
|
|
|
|
var cl sdkClient.Client
|
|
var prmInit sdkClient.PrmInit
|
|
prmInit.SetDefaultPrivateKey(c.prm.key)
|
|
prmInit.SetResponseInfoCallback(c.prm.responseInfoCallback)
|
|
|
|
cl.Init(prmInit)
|
|
|
|
var prmDial sdkClient.PrmDial
|
|
prmDial.SetServerURI(c.prm.address)
|
|
prmDial.SetTimeout(c.prm.dialTimeout)
|
|
prmDial.SetStreamTimeout(c.prm.streamTimeout)
|
|
prmDial.SetGRPCDialOptions(c.prm.dialOptions...)
|
|
|
|
if err := cl.Dial(ctx, prmDial); err != nil {
|
|
c.setUnhealthy()
|
|
return false, wasHealthy
|
|
}
|
|
|
|
c.clientMutex.Lock()
|
|
c.client = &cl
|
|
c.clientMutex.Unlock()
|
|
|
|
if _, err := cl.EndpointInfo(ctx, sdkClient.PrmEndpointInfo{}); err != nil {
|
|
c.setUnhealthy()
|
|
return false, wasHealthy
|
|
}
|
|
|
|
c.setHealthy()
|
|
return true, !wasHealthy
|
|
}
|
|
|
|
func (c *clientWrapper) getClient() (*sdkClient.Client, error) {
|
|
c.clientMutex.RLock()
|
|
defer c.clientMutex.RUnlock()
|
|
if c.isHealthy() {
|
|
return c.client, nil
|
|
}
|
|
return nil, errPoolClientUnhealthy
|
|
}
|
|
|
|
// balanceGet invokes sdkClient.BalanceGet parse response status to error and return result as is.
|
|
func (c *clientWrapper) balanceGet(ctx context.Context, prm PrmBalanceGet) (accounting.Decimal, error) {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return accounting.Decimal{}, err
|
|
}
|
|
|
|
var cliPrm sdkClient.PrmBalanceGet
|
|
cliPrm.SetAccount(prm.account)
|
|
|
|
start := time.Now()
|
|
res, err := cl.BalanceGet(ctx, cliPrm)
|
|
c.incRequests(time.Since(start), methodBalanceGet)
|
|
var st apistatus.Status
|
|
if res != nil {
|
|
st = res.Status()
|
|
}
|
|
if err = c.handleError(ctx, st, err); err != nil {
|
|
return accounting.Decimal{}, fmt.Errorf("balance get on client: %w", err)
|
|
}
|
|
|
|
return res.Amount(), nil
|
|
}
|
|
|
|
// containerPut invokes sdkClient.ContainerPut parse response status to error and return result as is.
|
|
// It also waits for the container to appear on the network.
|
|
func (c *clientWrapper) containerPut(ctx context.Context, prm PrmContainerPut) (cid.ID, error) {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return cid.ID{}, err
|
|
}
|
|
|
|
start := time.Now()
|
|
res, err := cl.ContainerPut(ctx, prm.prmClient)
|
|
c.incRequests(time.Since(start), methodContainerPut)
|
|
var st apistatus.Status
|
|
if res != nil {
|
|
st = res.Status()
|
|
}
|
|
if err = c.handleError(ctx, st, err); err != nil {
|
|
return cid.ID{}, fmt.Errorf("container put on client: %w", err)
|
|
}
|
|
|
|
if !prm.waitParamsSet {
|
|
prm.waitParams.setDefaults()
|
|
}
|
|
|
|
idCnr := res.ID()
|
|
|
|
err = waitForContainerPresence(ctx, c, idCnr, &prm.waitParams)
|
|
if err = c.handleError(ctx, nil, err); err != nil {
|
|
return cid.ID{}, fmt.Errorf("wait container presence on client: %w", err)
|
|
}
|
|
|
|
return idCnr, nil
|
|
}
|
|
|
|
// containerGet invokes sdkClient.ContainerGet parse response status to error and return result as is.
|
|
func (c *clientWrapper) containerGet(ctx context.Context, prm PrmContainerGet) (container.Container, error) {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return container.Container{}, err
|
|
}
|
|
|
|
var cliPrm sdkClient.PrmContainerGet
|
|
cliPrm.SetContainer(prm.cnrID)
|
|
|
|
start := time.Now()
|
|
res, err := cl.ContainerGet(ctx, cliPrm)
|
|
c.incRequests(time.Since(start), methodContainerGet)
|
|
var st apistatus.Status
|
|
if res != nil {
|
|
st = res.Status()
|
|
}
|
|
if err = c.handleError(ctx, st, err); err != nil {
|
|
return container.Container{}, fmt.Errorf("container get on client: %w", err)
|
|
}
|
|
|
|
return res.Container(), nil
|
|
}
|
|
|
|
// containerList invokes sdkClient.ContainerList parse response status to error and return result as is.
|
|
func (c *clientWrapper) containerList(ctx context.Context, prm PrmContainerList) ([]cid.ID, error) {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
var cliPrm sdkClient.PrmContainerList
|
|
cliPrm.SetAccount(prm.ownerID)
|
|
|
|
start := time.Now()
|
|
res, err := cl.ContainerList(ctx, cliPrm)
|
|
c.incRequests(time.Since(start), methodContainerList)
|
|
var st apistatus.Status
|
|
if res != nil {
|
|
st = res.Status()
|
|
}
|
|
if err = c.handleError(ctx, st, err); err != nil {
|
|
return nil, fmt.Errorf("container list on client: %w", err)
|
|
}
|
|
return res.Containers(), nil
|
|
}
|
|
|
|
// containerDelete invokes sdkClient.ContainerDelete parse response status to error.
|
|
// It also waits for the container to be removed from the network.
|
|
func (c *clientWrapper) containerDelete(ctx context.Context, prm PrmContainerDelete) error {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
var cliPrm sdkClient.PrmContainerDelete
|
|
cliPrm.SetContainer(prm.cnrID)
|
|
if prm.stokenSet {
|
|
cliPrm.WithinSession(prm.stoken)
|
|
}
|
|
|
|
start := time.Now()
|
|
res, err := cl.ContainerDelete(ctx, cliPrm)
|
|
c.incRequests(time.Since(start), methodContainerDelete)
|
|
var st apistatus.Status
|
|
if res != nil {
|
|
st = res.Status()
|
|
}
|
|
if err = c.handleError(ctx, st, err); err != nil {
|
|
return fmt.Errorf("container delete on client: %w", err)
|
|
}
|
|
|
|
if !prm.waitParamsSet {
|
|
prm.waitParams.setDefaults()
|
|
}
|
|
|
|
return waitForContainerRemoved(ctx, c, &prm.cnrID, &prm.waitParams)
|
|
}
|
|
|
|
// containerEACL invokes sdkClient.ContainerEACL parse response status to error and return result as is.
|
|
func (c *clientWrapper) containerEACL(ctx context.Context, prm PrmContainerEACL) (eacl.Table, error) {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return eacl.Table{}, err
|
|
}
|
|
|
|
var cliPrm sdkClient.PrmContainerEACL
|
|
cliPrm.SetContainer(prm.cnrID)
|
|
|
|
start := time.Now()
|
|
res, err := cl.ContainerEACL(ctx, cliPrm)
|
|
c.incRequests(time.Since(start), methodContainerEACL)
|
|
var st apistatus.Status
|
|
if res != nil {
|
|
st = res.Status()
|
|
}
|
|
if err = c.handleError(ctx, st, err); err != nil {
|
|
return eacl.Table{}, fmt.Errorf("get eacl on client: %w", err)
|
|
}
|
|
|
|
return res.Table(), nil
|
|
}
|
|
|
|
// containerSetEACL invokes sdkClient.ContainerSetEACL parse response status to error.
|
|
// It also waits for the EACL to appear on the network.
|
|
func (c *clientWrapper) containerSetEACL(ctx context.Context, prm PrmContainerSetEACL) error {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
var cliPrm sdkClient.PrmContainerSetEACL
|
|
cliPrm.SetTable(prm.table)
|
|
|
|
if prm.sessionSet {
|
|
cliPrm.WithinSession(prm.session)
|
|
}
|
|
|
|
start := time.Now()
|
|
res, err := cl.ContainerSetEACL(ctx, cliPrm)
|
|
c.incRequests(time.Since(start), methodContainerSetEACL)
|
|
var st apistatus.Status
|
|
if res != nil {
|
|
st = res.Status()
|
|
}
|
|
if err = c.handleError(ctx, st, err); err != nil {
|
|
return fmt.Errorf("set eacl on client: %w", err)
|
|
}
|
|
|
|
if !prm.waitParamsSet {
|
|
prm.waitParams.setDefaults()
|
|
}
|
|
|
|
var cIDp *cid.ID
|
|
if cID, set := prm.table.CID(); set {
|
|
cIDp = &cID
|
|
}
|
|
|
|
err = waitForEACLPresence(ctx, c, cIDp, &prm.table, &prm.waitParams)
|
|
if err = c.handleError(ctx, nil, err); err != nil {
|
|
return fmt.Errorf("wait eacl presence on client: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// endpointInfo invokes sdkClient.EndpointInfo parse response status to error and return result as is.
|
|
func (c *clientWrapper) endpointInfo(ctx context.Context, _ prmEndpointInfo) (netmap.NodeInfo, error) {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return netmap.NodeInfo{}, err
|
|
}
|
|
|
|
start := time.Now()
|
|
res, err := cl.EndpointInfo(ctx, sdkClient.PrmEndpointInfo{})
|
|
c.incRequests(time.Since(start), methodEndpointInfo)
|
|
var st apistatus.Status
|
|
if res != nil {
|
|
st = res.Status()
|
|
}
|
|
if err = c.handleError(ctx, st, err); err != nil {
|
|
return netmap.NodeInfo{}, fmt.Errorf("endpoint info on client: %w", err)
|
|
}
|
|
|
|
return res.NodeInfo(), nil
|
|
}
|
|
|
|
// networkInfo invokes sdkClient.NetworkInfo parse response status to error and return result as is.
|
|
func (c *clientWrapper) networkInfo(ctx context.Context, _ prmNetworkInfo) (netmap.NetworkInfo, error) {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return netmap.NetworkInfo{}, err
|
|
}
|
|
|
|
start := time.Now()
|
|
res, err := cl.NetworkInfo(ctx, sdkClient.PrmNetworkInfo{})
|
|
c.incRequests(time.Since(start), methodNetworkInfo)
|
|
var st apistatus.Status
|
|
if res != nil {
|
|
st = res.Status()
|
|
}
|
|
if err = c.handleError(ctx, st, err); err != nil {
|
|
return netmap.NetworkInfo{}, fmt.Errorf("network info on client: %w", err)
|
|
}
|
|
|
|
return res.Info(), nil
|
|
}
|
|
|
|
// objectPut writes object to FrostFS.
|
|
func (c *clientWrapper) objectPut(ctx context.Context, prm PrmObjectPut) (oid.ID, error) {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return oid.ID{}, err
|
|
}
|
|
|
|
var cliPrm sdkClient.PrmObjectPutInit
|
|
cliPrm.SetCopiesNumberByVectors(prm.copiesNumber)
|
|
if prm.stoken != nil {
|
|
cliPrm.WithinSession(*prm.stoken)
|
|
}
|
|
if prm.key != nil {
|
|
cliPrm.UseKey(*prm.key)
|
|
}
|
|
if prm.btoken != nil {
|
|
cliPrm.WithBearerToken(*prm.btoken)
|
|
}
|
|
|
|
start := time.Now()
|
|
wObj, err := cl.ObjectPutInit(ctx, cliPrm)
|
|
c.incRequests(time.Since(start), methodObjectPut)
|
|
if err = c.handleError(ctx, nil, err); err != nil {
|
|
return oid.ID{}, fmt.Errorf("init writing on API client: %w", err)
|
|
}
|
|
|
|
if wObj.WriteHeader(prm.hdr) {
|
|
sz := prm.hdr.PayloadSize()
|
|
|
|
if data := prm.hdr.Payload(); len(data) > 0 {
|
|
if prm.payload != nil {
|
|
prm.payload = io.MultiReader(bytes.NewReader(data), prm.payload)
|
|
} else {
|
|
prm.payload = bytes.NewReader(data)
|
|
sz = uint64(len(data))
|
|
}
|
|
}
|
|
|
|
if prm.payload != nil {
|
|
const defaultBufferSizePut = 3 << 20 // configure?
|
|
|
|
if sz == 0 || sz > defaultBufferSizePut {
|
|
sz = defaultBufferSizePut
|
|
}
|
|
|
|
buf := make([]byte, sz)
|
|
|
|
var n int
|
|
|
|
for {
|
|
n, err = prm.payload.Read(buf)
|
|
if n > 0 {
|
|
start = time.Now()
|
|
successWrite := wObj.WritePayloadChunk(buf[:n])
|
|
c.incRequests(time.Since(start), methodObjectPut)
|
|
if !successWrite {
|
|
break
|
|
}
|
|
|
|
continue
|
|
}
|
|
|
|
if errors.Is(err, io.EOF) {
|
|
break
|
|
}
|
|
|
|
return oid.ID{}, fmt.Errorf("read payload: %w", c.handleError(ctx, nil, err))
|
|
}
|
|
}
|
|
}
|
|
|
|
res, err := wObj.Close()
|
|
var st apistatus.Status
|
|
if res != nil {
|
|
st = res.Status()
|
|
}
|
|
if err = c.handleError(ctx, st, err); err != nil { // here err already carries both status and client errors
|
|
return oid.ID{}, fmt.Errorf("client failure: %w", err)
|
|
}
|
|
|
|
return res.StoredObjectID(), nil
|
|
}
|
|
|
|
// objectDelete invokes sdkClient.ObjectDelete parse response status to error.
|
|
func (c *clientWrapper) objectDelete(ctx context.Context, prm PrmObjectDelete) error {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
var cliPrm sdkClient.PrmObjectDelete
|
|
cliPrm.FromContainer(prm.addr.Container())
|
|
cliPrm.ByID(prm.addr.Object())
|
|
|
|
if prm.stoken != nil {
|
|
cliPrm.WithinSession(*prm.stoken)
|
|
}
|
|
|
|
if prm.btoken != nil {
|
|
cliPrm.WithBearerToken(*prm.btoken)
|
|
}
|
|
|
|
if prm.key != nil {
|
|
cliPrm.UseKey(*prm.key)
|
|
}
|
|
|
|
start := time.Now()
|
|
res, err := cl.ObjectDelete(ctx, cliPrm)
|
|
c.incRequests(time.Since(start), methodObjectDelete)
|
|
var st apistatus.Status
|
|
if res != nil {
|
|
st = res.Status()
|
|
}
|
|
if err = c.handleError(ctx, st, err); err != nil {
|
|
return fmt.Errorf("delete object on client: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// objectGet returns reader for object.
|
|
func (c *clientWrapper) objectGet(ctx context.Context, prm PrmObjectGet) (ResGetObject, error) {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return ResGetObject{}, err
|
|
}
|
|
|
|
var cliPrm sdkClient.PrmObjectGet
|
|
cliPrm.FromContainer(prm.addr.Container())
|
|
cliPrm.ByID(prm.addr.Object())
|
|
|
|
if prm.stoken != nil {
|
|
cliPrm.WithinSession(*prm.stoken)
|
|
}
|
|
|
|
if prm.btoken != nil {
|
|
cliPrm.WithBearerToken(*prm.btoken)
|
|
}
|
|
|
|
if prm.key != nil {
|
|
cliPrm.UseKey(*prm.key)
|
|
}
|
|
|
|
var res ResGetObject
|
|
|
|
rObj, err := cl.ObjectGetInit(ctx, cliPrm)
|
|
if err = c.handleError(ctx, nil, err); err != nil {
|
|
return ResGetObject{}, fmt.Errorf("init object reading on client: %w", err)
|
|
}
|
|
|
|
start := time.Now()
|
|
successReadHeader := rObj.ReadHeader(&res.Header)
|
|
c.incRequests(time.Since(start), methodObjectGet)
|
|
if !successReadHeader {
|
|
rObjRes, err := rObj.Close()
|
|
var st apistatus.Status
|
|
if rObjRes != nil {
|
|
st = rObjRes.Status()
|
|
}
|
|
err = c.handleError(ctx, st, err)
|
|
return res, fmt.Errorf("read header: %w", err)
|
|
}
|
|
|
|
res.Payload = &objectReadCloser{
|
|
reader: rObj,
|
|
elapsedTimeCallback: func(elapsed time.Duration) {
|
|
c.incRequests(elapsed, methodObjectGet)
|
|
},
|
|
}
|
|
|
|
return res, nil
|
|
}
|
|
|
|
// objectHead invokes sdkClient.ObjectHead parse response status to error and return result as is.
|
|
func (c *clientWrapper) objectHead(ctx context.Context, prm PrmObjectHead) (object.Object, error) {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return object.Object{}, err
|
|
}
|
|
|
|
var cliPrm sdkClient.PrmObjectHead
|
|
cliPrm.FromContainer(prm.addr.Container())
|
|
cliPrm.ByID(prm.addr.Object())
|
|
if prm.raw {
|
|
cliPrm.MarkRaw()
|
|
}
|
|
|
|
if prm.stoken != nil {
|
|
cliPrm.WithinSession(*prm.stoken)
|
|
}
|
|
|
|
if prm.btoken != nil {
|
|
cliPrm.WithBearerToken(*prm.btoken)
|
|
}
|
|
|
|
if prm.key != nil {
|
|
cliPrm.UseKey(*prm.key)
|
|
}
|
|
|
|
var obj object.Object
|
|
|
|
start := time.Now()
|
|
res, err := cl.ObjectHead(ctx, cliPrm)
|
|
c.incRequests(time.Since(start), methodObjectHead)
|
|
var st apistatus.Status
|
|
if res != nil {
|
|
st = res.Status()
|
|
}
|
|
if err = c.handleError(ctx, st, err); err != nil {
|
|
return obj, fmt.Errorf("read object header via client: %w", err)
|
|
}
|
|
if !res.ReadHeader(&obj) {
|
|
return obj, errors.New("missing object header in response")
|
|
}
|
|
|
|
return obj, nil
|
|
}
|
|
|
|
// objectRange returns object range reader.
|
|
func (c *clientWrapper) objectRange(ctx context.Context, prm PrmObjectRange) (ResObjectRange, error) {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return ResObjectRange{}, err
|
|
}
|
|
|
|
var cliPrm sdkClient.PrmObjectRange
|
|
cliPrm.FromContainer(prm.addr.Container())
|
|
cliPrm.ByID(prm.addr.Object())
|
|
cliPrm.SetOffset(prm.off)
|
|
cliPrm.SetLength(prm.ln)
|
|
|
|
if prm.stoken != nil {
|
|
cliPrm.WithinSession(*prm.stoken)
|
|
}
|
|
|
|
if prm.btoken != nil {
|
|
cliPrm.WithBearerToken(*prm.btoken)
|
|
}
|
|
|
|
if prm.key != nil {
|
|
cliPrm.UseKey(*prm.key)
|
|
}
|
|
|
|
start := time.Now()
|
|
res, err := cl.ObjectRangeInit(ctx, cliPrm)
|
|
c.incRequests(time.Since(start), methodObjectRange)
|
|
if err = c.handleError(ctx, nil, err); err != nil {
|
|
return ResObjectRange{}, fmt.Errorf("init payload range reading on client: %w", err)
|
|
}
|
|
|
|
return ResObjectRange{
|
|
payload: res,
|
|
elapsedTimeCallback: func(elapsed time.Duration) {
|
|
c.incRequests(elapsed, methodObjectRange)
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
// objectSearch invokes sdkClient.ObjectSearchInit parse response status to error and return result as is.
|
|
func (c *clientWrapper) objectSearch(ctx context.Context, prm PrmObjectSearch) (ResObjectSearch, error) {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return ResObjectSearch{}, err
|
|
}
|
|
|
|
var cliPrm sdkClient.PrmObjectSearch
|
|
|
|
cliPrm.InContainer(prm.cnrID)
|
|
cliPrm.SetFilters(prm.filters)
|
|
|
|
if prm.stoken != nil {
|
|
cliPrm.WithinSession(*prm.stoken)
|
|
}
|
|
|
|
if prm.btoken != nil {
|
|
cliPrm.WithBearerToken(*prm.btoken)
|
|
}
|
|
|
|
if prm.key != nil {
|
|
cliPrm.UseKey(*prm.key)
|
|
}
|
|
|
|
res, err := cl.ObjectSearchInit(ctx, cliPrm)
|
|
if err = c.handleError(ctx, nil, err); err != nil {
|
|
return ResObjectSearch{}, fmt.Errorf("init object searching on client: %w", err)
|
|
}
|
|
|
|
return ResObjectSearch{r: res}, nil
|
|
}
|
|
|
|
// sessionCreate invokes sdkClient.SessionCreate parse response status to error and return result as is.
|
|
func (c *clientWrapper) sessionCreate(ctx context.Context, prm prmCreateSession) (resCreateSession, error) {
|
|
cl, err := c.getClient()
|
|
if err != nil {
|
|
return resCreateSession{}, err
|
|
}
|
|
|
|
var cliPrm sdkClient.PrmSessionCreate
|
|
cliPrm.SetExp(prm.exp)
|
|
cliPrm.UseKey(prm.key)
|
|
|
|
start := time.Now()
|
|
res, err := cl.SessionCreate(ctx, cliPrm)
|
|
c.incRequests(time.Since(start), methodSessionCreate)
|
|
var st apistatus.Status
|
|
if res != nil {
|
|
st = res.Status()
|
|
}
|
|
if err = c.handleError(ctx, st, err); err != nil {
|
|
return resCreateSession{}, fmt.Errorf("session creation on client: %w", err)
|
|
}
|
|
|
|
return resCreateSession{
|
|
id: res.ID(),
|
|
sessionKey: res.PublicKey(),
|
|
}, nil
|
|
}
|
|
|
|
func (c *clientStatusMonitor) isHealthy() bool {
|
|
return c.healthy.Load()
|
|
}
|
|
|
|
func (c *clientStatusMonitor) setHealthy() {
|
|
c.healthy.Store(true)
|
|
}
|
|
|
|
func (c *clientStatusMonitor) setUnhealthy() {
|
|
c.healthy.Store(false)
|
|
}
|
|
|
|
func (c *clientStatusMonitor) address() string {
|
|
return c.addr
|
|
}
|
|
|
|
func (c *clientStatusMonitor) incErrorRate() {
|
|
c.mu.Lock()
|
|
c.currentErrorCount++
|
|
c.overallErrorCount++
|
|
|
|
thresholdReached := c.currentErrorCount >= c.errorThreshold
|
|
if thresholdReached {
|
|
c.setUnhealthy()
|
|
c.currentErrorCount = 0
|
|
}
|
|
c.mu.Unlock()
|
|
|
|
if thresholdReached && c.logger != nil {
|
|
c.logger.Warn("error threshold reached",
|
|
zap.String("address", c.addr), zap.Uint32("threshold", c.errorThreshold))
|
|
}
|
|
}
|
|
|
|
func (c *clientStatusMonitor) currentErrorRate() uint32 {
|
|
c.mu.RLock()
|
|
defer c.mu.RUnlock()
|
|
return c.currentErrorCount
|
|
}
|
|
|
|
func (c *clientStatusMonitor) overallErrorRate() uint64 {
|
|
c.mu.RLock()
|
|
defer c.mu.RUnlock()
|
|
return c.overallErrorCount
|
|
}
|
|
|
|
func (c *clientStatusMonitor) methodsStatus() []statusSnapshot {
|
|
result := make([]statusSnapshot, len(c.methods))
|
|
for i, val := range c.methods {
|
|
result[i] = val.snapshot()
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
func (c *clientWrapper) incRequests(elapsed time.Duration, method MethodIndex) {
|
|
methodStat := c.methods[method]
|
|
methodStat.incRequests(elapsed)
|
|
if c.prm.poolRequestInfoCallback != nil {
|
|
c.prm.poolRequestInfoCallback(RequestInfo{
|
|
Address: c.prm.address,
|
|
Method: method,
|
|
Elapsed: elapsed,
|
|
})
|
|
}
|
|
}
|
|
|
|
func (c *clientStatusMonitor) handleError(ctx context.Context, st apistatus.Status, err error) error {
|
|
if err != nil {
|
|
if needCountError(ctx, err) {
|
|
c.incErrorRate()
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
err = apistatus.ErrFromStatus(st)
|
|
switch err.(type) {
|
|
case apistatus.ServerInternal, *apistatus.ServerInternal,
|
|
apistatus.WrongMagicNumber, *apistatus.WrongMagicNumber,
|
|
apistatus.SignatureVerification, *apistatus.SignatureVerification,
|
|
apistatus.NodeUnderMaintenance, *apistatus.NodeUnderMaintenance:
|
|
c.incErrorRate()
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
func needCountError(ctx context.Context, err error) bool {
|
|
// non-status logic error that could be returned
|
|
// from the SDK client; should not be considered
|
|
// as a connection error
|
|
var siErr *object.SplitInfoError
|
|
if errors.As(err, &siErr) {
|
|
return false
|
|
}
|
|
|
|
if errors.Is(ctx.Err(), context.Canceled) {
|
|
return false
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
// clientBuilder is a type alias of client constructors which open connection
|
|
// to the given endpoint.
|
|
type clientBuilder = func(endpoint string) client
|
|
|
|
// RequestInfo groups info about pool request.
|
|
type RequestInfo struct {
|
|
Address string
|
|
Method MethodIndex
|
|
Elapsed time.Duration
|
|
}
|
|
|
|
// InitParameters contains values used to initialize connection Pool.
|
|
type InitParameters struct {
|
|
key *ecdsa.PrivateKey
|
|
logger *zap.Logger
|
|
nodeDialTimeout time.Duration
|
|
nodeStreamTimeout time.Duration
|
|
healthcheckTimeout time.Duration
|
|
clientRebalanceInterval time.Duration
|
|
sessionExpirationDuration uint64
|
|
errorThreshold uint32
|
|
nodeParams []NodeParam
|
|
requestCallback func(RequestInfo)
|
|
dialOptions []grpc.DialOption
|
|
|
|
clientBuilder clientBuilder
|
|
}
|
|
|
|
// SetKey specifies default key to be used for the protocol communication by default.
|
|
func (x *InitParameters) SetKey(key *ecdsa.PrivateKey) {
|
|
x.key = key
|
|
}
|
|
|
|
// SetLogger specifies logger.
|
|
func (x *InitParameters) SetLogger(logger *zap.Logger) {
|
|
x.logger = logger
|
|
}
|
|
|
|
// SetNodeDialTimeout specifies the timeout for connection to be established.
|
|
func (x *InitParameters) SetNodeDialTimeout(timeout time.Duration) {
|
|
x.nodeDialTimeout = timeout
|
|
}
|
|
|
|
// SetNodeStreamTimeout specifies the timeout for individual operations in streaming RPC.
|
|
func (x *InitParameters) SetNodeStreamTimeout(timeout time.Duration) {
|
|
x.nodeStreamTimeout = timeout
|
|
}
|
|
|
|
// SetHealthcheckTimeout specifies the timeout for request to node to decide if it is alive.
|
|
//
|
|
// See also Pool.Dial.
|
|
func (x *InitParameters) SetHealthcheckTimeout(timeout time.Duration) {
|
|
x.healthcheckTimeout = timeout
|
|
}
|
|
|
|
// SetClientRebalanceInterval specifies the interval for updating nodes health status.
|
|
//
|
|
// See also Pool.Dial.
|
|
func (x *InitParameters) SetClientRebalanceInterval(interval time.Duration) {
|
|
x.clientRebalanceInterval = interval
|
|
}
|
|
|
|
// SetSessionExpirationDuration specifies the session token lifetime in epochs.
|
|
func (x *InitParameters) SetSessionExpirationDuration(expirationDuration uint64) {
|
|
x.sessionExpirationDuration = expirationDuration
|
|
}
|
|
|
|
// SetErrorThreshold specifies the number of errors on connection after which node is considered as unhealthy.
|
|
func (x *InitParameters) SetErrorThreshold(threshold uint32) {
|
|
x.errorThreshold = threshold
|
|
}
|
|
|
|
// SetRequestCallback makes the pool client to pass RequestInfo for each
|
|
// request to f. Nil (default) means ignore RequestInfo.
|
|
func (x *InitParameters) SetRequestCallback(f func(RequestInfo)) {
|
|
x.requestCallback = f
|
|
}
|
|
|
|
// AddNode append information about the node to which you want to connect.
|
|
func (x *InitParameters) AddNode(nodeParam NodeParam) {
|
|
x.nodeParams = append(x.nodeParams, nodeParam)
|
|
}
|
|
|
|
// SetGRPCDialOptions sets the gRPC dial options for new gRPC client connection.
|
|
func (x *InitParameters) SetGRPCDialOptions(opts ...grpc.DialOption) {
|
|
x.dialOptions = opts
|
|
}
|
|
|
|
// setClientBuilder sets clientBuilder used for client construction.
|
|
// Wraps setClientBuilderContext without a context.
|
|
func (x *InitParameters) setClientBuilder(builder clientBuilder) {
|
|
x.clientBuilder = builder
|
|
}
|
|
|
|
// isMissingClientBuilder checks if client constructor was not specified.
|
|
func (x *InitParameters) isMissingClientBuilder() bool {
|
|
return x.clientBuilder == nil
|
|
}
|
|
|
|
type rebalanceParameters struct {
|
|
nodesParams []*nodesParam
|
|
nodeRequestTimeout time.Duration
|
|
clientRebalanceInterval time.Duration
|
|
sessionExpirationDuration uint64
|
|
}
|
|
|
|
type nodesParam struct {
|
|
priority int
|
|
addresses []string
|
|
weights []float64
|
|
}
|
|
|
|
// NodeParam groups parameters of remote node.
|
|
type NodeParam struct {
|
|
priority int
|
|
address string
|
|
weight float64
|
|
}
|
|
|
|
// NewNodeParam creates NodeParam using parameters.
|
|
func NewNodeParam(priority int, address string, weight float64) (prm NodeParam) {
|
|
prm.SetPriority(priority)
|
|
prm.SetAddress(address)
|
|
prm.SetWeight(weight)
|
|
|
|
return
|
|
}
|
|
|
|
// SetPriority specifies priority of the node.
|
|
// Negative value is allowed. In the result node groups
|
|
// with the same priority will be sorted by descent.
|
|
func (x *NodeParam) SetPriority(priority int) {
|
|
x.priority = priority
|
|
}
|
|
|
|
// Priority returns priority of the node.
|
|
func (x *NodeParam) Priority() int {
|
|
return x.priority
|
|
}
|
|
|
|
// SetAddress specifies address of the node.
|
|
func (x *NodeParam) SetAddress(address string) {
|
|
x.address = address
|
|
}
|
|
|
|
// Address returns address of the node.
|
|
func (x *NodeParam) Address() string {
|
|
return x.address
|
|
}
|
|
|
|
// SetWeight specifies weight of the node.
|
|
func (x *NodeParam) SetWeight(weight float64) {
|
|
x.weight = weight
|
|
}
|
|
|
|
// Weight returns weight of the node.
|
|
func (x *NodeParam) Weight() float64 {
|
|
return x.weight
|
|
}
|
|
|
|
// WaitParams contains parameters used in polling is a something applied on FrostFS network.
|
|
type WaitParams struct {
|
|
timeout time.Duration
|
|
pollInterval time.Duration
|
|
}
|
|
|
|
// SetTimeout specifies the time to wait for the operation to complete.
|
|
func (x *WaitParams) SetTimeout(timeout time.Duration) {
|
|
x.timeout = timeout
|
|
}
|
|
|
|
// SetPollInterval specifies the interval, once it will check the completion of the operation.
|
|
func (x *WaitParams) SetPollInterval(tick time.Duration) {
|
|
x.pollInterval = tick
|
|
}
|
|
|
|
func (x *WaitParams) setDefaults() {
|
|
x.timeout = 120 * time.Second
|
|
x.pollInterval = 5 * time.Second
|
|
}
|
|
|
|
// checkForPositive panics if any of the wait params isn't positive.
|
|
func (x *WaitParams) checkForPositive() {
|
|
if x.timeout <= 0 || x.pollInterval <= 0 {
|
|
panic("all wait params must be positive")
|
|
}
|
|
}
|
|
|
|
type prmContext struct {
|
|
defaultSession bool
|
|
verb session.ObjectVerb
|
|
cnr cid.ID
|
|
|
|
objSet bool
|
|
objs []oid.ID
|
|
}
|
|
|
|
func (x *prmContext) useDefaultSession() {
|
|
x.defaultSession = true
|
|
}
|
|
|
|
func (x *prmContext) useContainer(cnr cid.ID) {
|
|
x.cnr = cnr
|
|
}
|
|
|
|
func (x *prmContext) useObjects(ids []oid.ID) {
|
|
x.objs = ids
|
|
x.objSet = true
|
|
}
|
|
|
|
func (x *prmContext) useAddress(addr oid.Address) {
|
|
x.cnr = addr.Container()
|
|
x.objs = []oid.ID{addr.Object()}
|
|
x.objSet = true
|
|
}
|
|
|
|
func (x *prmContext) useVerb(verb session.ObjectVerb) {
|
|
x.verb = verb
|
|
}
|
|
|
|
type prmCommon struct {
|
|
key *ecdsa.PrivateKey
|
|
btoken *bearer.Token
|
|
stoken *session.Object
|
|
}
|
|
|
|
// UseKey specifies private key to sign the requests.
|
|
// If key is not provided, then Pool default key is used.
|
|
func (x *prmCommon) UseKey(key *ecdsa.PrivateKey) {
|
|
x.key = key
|
|
}
|
|
|
|
// UseBearer attaches bearer token to be used for the operation.
|
|
func (x *prmCommon) UseBearer(token bearer.Token) {
|
|
x.btoken = &token
|
|
}
|
|
|
|
// UseSession specifies session within which operation should be performed.
|
|
func (x *prmCommon) UseSession(token session.Object) {
|
|
x.stoken = &token
|
|
}
|
|
|
|
// PrmObjectPut groups parameters of PutObject operation.
|
|
type PrmObjectPut struct {
|
|
prmCommon
|
|
|
|
hdr object.Object
|
|
|
|
payload io.Reader
|
|
|
|
copiesNumber []uint32
|
|
}
|
|
|
|
// SetHeader specifies header of the object.
|
|
func (x *PrmObjectPut) SetHeader(hdr object.Object) {
|
|
x.hdr = hdr
|
|
}
|
|
|
|
// SetPayload specifies payload of the object.
|
|
func (x *PrmObjectPut) SetPayload(payload io.Reader) {
|
|
x.payload = payload
|
|
}
|
|
|
|
// SetCopiesNumber sets number of object copies that is enough to consider put successful.
|
|
// Zero means using default behavior.
|
|
func (x *PrmObjectPut) SetCopiesNumber(copiesNumber uint32) {
|
|
x.copiesNumber = []uint32{copiesNumber}
|
|
}
|
|
|
|
// SetCopiesNumberVector sets number of object copies that is enough to consider put successful, provided as array.
|
|
// Nil/empty vector means using default behavior.
|
|
func (x *PrmObjectPut) SetCopiesNumberVector(copiesNumber []uint32) {
|
|
x.copiesNumber = copiesNumber
|
|
}
|
|
|
|
// PrmObjectDelete groups parameters of DeleteObject operation.
|
|
type PrmObjectDelete struct {
|
|
prmCommon
|
|
|
|
addr oid.Address
|
|
}
|
|
|
|
// SetAddress specifies FrostFS address of the object.
|
|
func (x *PrmObjectDelete) SetAddress(addr oid.Address) {
|
|
x.addr = addr
|
|
}
|
|
|
|
// PrmObjectGet groups parameters of GetObject operation.
|
|
type PrmObjectGet struct {
|
|
prmCommon
|
|
|
|
addr oid.Address
|
|
}
|
|
|
|
// SetAddress specifies FrostFS address of the object.
|
|
func (x *PrmObjectGet) SetAddress(addr oid.Address) {
|
|
x.addr = addr
|
|
}
|
|
|
|
// PrmObjectHead groups parameters of HeadObject operation.
|
|
type PrmObjectHead struct {
|
|
prmCommon
|
|
|
|
addr oid.Address
|
|
raw bool
|
|
}
|
|
|
|
// SetAddress specifies FrostFS address of the object.
|
|
func (x *PrmObjectHead) SetAddress(addr oid.Address) {
|
|
x.addr = addr
|
|
}
|
|
|
|
// MarkRaw marks an intent to read physically stored object.
|
|
func (x *PrmObjectHead) MarkRaw() {
|
|
x.raw = true
|
|
}
|
|
|
|
// PrmObjectRange groups parameters of RangeObject operation.
|
|
type PrmObjectRange struct {
|
|
prmCommon
|
|
|
|
addr oid.Address
|
|
off, ln uint64
|
|
}
|
|
|
|
// SetAddress specifies FrostFS address of the object.
|
|
func (x *PrmObjectRange) SetAddress(addr oid.Address) {
|
|
x.addr = addr
|
|
}
|
|
|
|
// SetOffset sets offset of the payload range to be read.
|
|
func (x *PrmObjectRange) SetOffset(offset uint64) {
|
|
x.off = offset
|
|
}
|
|
|
|
// SetLength sets length of the payload range to be read.
|
|
func (x *PrmObjectRange) SetLength(length uint64) {
|
|
x.ln = length
|
|
}
|
|
|
|
// PrmObjectSearch groups parameters of SearchObjects operation.
|
|
type PrmObjectSearch struct {
|
|
prmCommon
|
|
|
|
cnrID cid.ID
|
|
filters object.SearchFilters
|
|
}
|
|
|
|
// SetContainerID specifies the container in which to look for objects.
|
|
func (x *PrmObjectSearch) SetContainerID(cnrID cid.ID) {
|
|
x.cnrID = cnrID
|
|
}
|
|
|
|
// SetFilters specifies filters by which to select objects.
|
|
func (x *PrmObjectSearch) SetFilters(filters object.SearchFilters) {
|
|
x.filters = filters
|
|
}
|
|
|
|
// PrmContainerPut groups parameters of PutContainer operation.
|
|
type PrmContainerPut struct {
|
|
prmClient sdkClient.PrmContainerPut
|
|
|
|
waitParams WaitParams
|
|
waitParamsSet bool
|
|
}
|
|
|
|
// SetContainer container structure to be used as a parameter of the base
|
|
// client's operation.
|
|
//
|
|
// See git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client.PrmContainerPut.SetContainer.
|
|
func (x *PrmContainerPut) SetContainer(cnr container.Container) {
|
|
x.prmClient.SetContainer(cnr)
|
|
}
|
|
|
|
// WithinSession specifies session to be used as a parameter of the base
|
|
// client's operation.
|
|
//
|
|
// See git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client.PrmContainerPut.WithinSession.
|
|
func (x *PrmContainerPut) WithinSession(s session.Container) {
|
|
x.prmClient.WithinSession(s)
|
|
}
|
|
|
|
// SetWaitParams specifies timeout params to complete operation.
|
|
// If not provided the default one will be used.
|
|
// Panics if any of the wait params isn't positive.
|
|
func (x *PrmContainerPut) SetWaitParams(waitParams WaitParams) {
|
|
waitParams.checkForPositive()
|
|
x.waitParams = waitParams
|
|
x.waitParamsSet = true
|
|
}
|
|
|
|
// PrmContainerGet groups parameters of GetContainer operation.
|
|
type PrmContainerGet struct {
|
|
cnrID cid.ID
|
|
}
|
|
|
|
// SetContainerID specifies identifier of the container to be read.
|
|
func (x *PrmContainerGet) SetContainerID(cnrID cid.ID) {
|
|
x.cnrID = cnrID
|
|
}
|
|
|
|
// PrmContainerList groups parameters of ListContainers operation.
|
|
type PrmContainerList struct {
|
|
ownerID user.ID
|
|
}
|
|
|
|
// SetOwnerID specifies identifier of the FrostFS account to list the containers.
|
|
func (x *PrmContainerList) SetOwnerID(ownerID user.ID) {
|
|
x.ownerID = ownerID
|
|
}
|
|
|
|
// PrmContainerDelete groups parameters of DeleteContainer operation.
|
|
type PrmContainerDelete struct {
|
|
cnrID cid.ID
|
|
|
|
stoken session.Container
|
|
stokenSet bool
|
|
|
|
waitParams WaitParams
|
|
waitParamsSet bool
|
|
}
|
|
|
|
// SetContainerID specifies identifier of the FrostFS container to be removed.
|
|
func (x *PrmContainerDelete) SetContainerID(cnrID cid.ID) {
|
|
x.cnrID = cnrID
|
|
}
|
|
|
|
// SetSessionToken specifies session within which operation should be performed.
|
|
func (x *PrmContainerDelete) SetSessionToken(token session.Container) {
|
|
x.stoken = token
|
|
x.stokenSet = true
|
|
}
|
|
|
|
// SetWaitParams specifies timeout params to complete operation.
|
|
// If not provided the default one will be used.
|
|
// Panics if any of the wait params isn't positive.
|
|
func (x *PrmContainerDelete) SetWaitParams(waitParams WaitParams) {
|
|
waitParams.checkForPositive()
|
|
x.waitParams = waitParams
|
|
x.waitParamsSet = true
|
|
}
|
|
|
|
// PrmContainerEACL groups parameters of GetEACL operation.
|
|
type PrmContainerEACL struct {
|
|
cnrID cid.ID
|
|
}
|
|
|
|
// SetContainerID specifies identifier of the FrostFS container to read the eACL table.
|
|
func (x *PrmContainerEACL) SetContainerID(cnrID cid.ID) {
|
|
x.cnrID = cnrID
|
|
}
|
|
|
|
// PrmContainerSetEACL groups parameters of SetEACL operation.
|
|
type PrmContainerSetEACL struct {
|
|
table eacl.Table
|
|
|
|
sessionSet bool
|
|
session session.Container
|
|
|
|
waitParams WaitParams
|
|
waitParamsSet bool
|
|
}
|
|
|
|
// SetTable sets structure of container's extended ACL to be used as a
|
|
// parameter of the base client's operation.
|
|
//
|
|
// See git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client.PrmContainerSetEACL.SetTable.
|
|
func (x *PrmContainerSetEACL) SetTable(table eacl.Table) {
|
|
x.table = table
|
|
}
|
|
|
|
// WithinSession specifies session to be used as a parameter of the base
|
|
// client's operation.
|
|
//
|
|
// See git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client.PrmContainerSetEACL.WithinSession.
|
|
func (x *PrmContainerSetEACL) WithinSession(s session.Container) {
|
|
x.session = s
|
|
x.sessionSet = true
|
|
}
|
|
|
|
// SetWaitParams specifies timeout params to complete operation.
|
|
// If not provided the default one will be used.
|
|
// Panics if any of the wait params isn't positive.
|
|
func (x *PrmContainerSetEACL) SetWaitParams(waitParams WaitParams) {
|
|
waitParams.checkForPositive()
|
|
x.waitParams = waitParams
|
|
x.waitParamsSet = true
|
|
}
|
|
|
|
// PrmBalanceGet groups parameters of Balance operation.
|
|
type PrmBalanceGet struct {
|
|
account user.ID
|
|
}
|
|
|
|
// SetAccount specifies identifier of the FrostFS account for which the balance is requested.
|
|
func (x *PrmBalanceGet) SetAccount(id user.ID) {
|
|
x.account = id
|
|
}
|
|
|
|
// prmEndpointInfo groups parameters of sessionCreate operation.
|
|
type prmCreateSession struct {
|
|
exp uint64
|
|
key ecdsa.PrivateKey
|
|
}
|
|
|
|
// setExp sets number of the last FrostFS epoch in the lifetime of the session after which it will be expired.
|
|
func (x *prmCreateSession) setExp(exp uint64) {
|
|
x.exp = exp
|
|
}
|
|
|
|
// useKey specifies owner private key for session token.
|
|
// If key is not provided, then Pool default key is used.
|
|
func (x *prmCreateSession) useKey(key ecdsa.PrivateKey) {
|
|
x.key = key
|
|
}
|
|
|
|
// prmEndpointInfo groups parameters of endpointInfo operation.
|
|
type prmEndpointInfo struct{}
|
|
|
|
// prmNetworkInfo groups parameters of networkInfo operation.
|
|
type prmNetworkInfo struct{}
|
|
|
|
// resCreateSession groups resulting values of sessionCreate operation.
|
|
type resCreateSession struct {
|
|
id []byte
|
|
|
|
sessionKey []byte
|
|
}
|
|
|
|
// Pool represents virtual connection to the FrostFS network to communicate
|
|
// with multiple FrostFS servers without thinking about switching between servers
|
|
// due to load balancing proportions or their unavailability.
|
|
// It is designed to provide a convenient abstraction from the multiple sdkClient.client types.
|
|
//
|
|
// Pool can be created and initialized using NewPool function.
|
|
// Before executing the FrostFS operations using the Pool, connection to the
|
|
// servers MUST BE correctly established (see Dial method).
|
|
// Using the Pool before connecting have been established can lead to a panic.
|
|
// After the work, the Pool SHOULD BE closed (see Close method): it frees internal
|
|
// and system resources which were allocated for the period of work of the Pool.
|
|
// Calling Dial/Close methods during the communication process step strongly discouraged
|
|
// as it leads to undefined behavior.
|
|
//
|
|
// Each method which produces a FrostFS API call may return an error.
|
|
// Status of underlying server response is casted to built-in error instance.
|
|
// Certain statuses can be checked using `sdkClient` and standard `errors` packages.
|
|
// Note that package provides some helper functions to work with status returns
|
|
// (e.g. sdkClient.IsErrContainerNotFound, sdkClient.IsErrObjectNotFound).
|
|
//
|
|
// See pool package overview to get some examples.
|
|
type Pool struct {
|
|
innerPools []*innerPool
|
|
key *ecdsa.PrivateKey
|
|
cancel context.CancelFunc
|
|
closedCh chan struct{}
|
|
cache *sessionCache
|
|
stokenDuration uint64
|
|
rebalanceParams rebalanceParameters
|
|
clientBuilder clientBuilder
|
|
logger *zap.Logger
|
|
}
|
|
|
|
type innerPool struct {
|
|
lock sync.RWMutex
|
|
sampler *sampler
|
|
clients []client
|
|
}
|
|
|
|
const (
|
|
defaultSessionTokenExpirationDuration = 100 // in blocks
|
|
defaultErrorThreshold = 100
|
|
|
|
defaultRebalanceInterval = 15 * time.Second
|
|
defaultHealthcheckTimeout = 4 * time.Second
|
|
defaultDialTimeout = 5 * time.Second
|
|
defaultStreamTimeout = 10 * time.Second
|
|
)
|
|
|
|
// NewPool creates connection pool using parameters.
|
|
func NewPool(options InitParameters) (*Pool, error) {
|
|
if options.key == nil {
|
|
return nil, fmt.Errorf("missed required parameter 'Key'")
|
|
}
|
|
|
|
nodesParams, err := adjustNodeParams(options.nodeParams)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cache, err := newCache()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("couldn't create cache: %w", err)
|
|
}
|
|
|
|
fillDefaultInitParams(&options, cache)
|
|
|
|
pool := &Pool{
|
|
key: options.key,
|
|
cache: cache,
|
|
logger: options.logger,
|
|
stokenDuration: options.sessionExpirationDuration,
|
|
rebalanceParams: rebalanceParameters{
|
|
nodesParams: nodesParams,
|
|
nodeRequestTimeout: options.healthcheckTimeout,
|
|
clientRebalanceInterval: options.clientRebalanceInterval,
|
|
sessionExpirationDuration: options.sessionExpirationDuration,
|
|
},
|
|
clientBuilder: options.clientBuilder,
|
|
}
|
|
|
|
return pool, nil
|
|
}
|
|
|
|
// Dial establishes a connection to the servers from the FrostFS network.
|
|
// It also starts a routine that checks the health of the nodes and
|
|
// updates the weights of the nodes for balancing.
|
|
// Returns an error describing failure reason.
|
|
//
|
|
// If failed, the Pool SHOULD NOT be used.
|
|
//
|
|
// See also InitParameters.SetClientRebalanceInterval.
|
|
func (p *Pool) Dial(ctx context.Context) error {
|
|
inner := make([]*innerPool, len(p.rebalanceParams.nodesParams))
|
|
var atLeastOneHealthy bool
|
|
|
|
for i, params := range p.rebalanceParams.nodesParams {
|
|
clients := make([]client, len(params.weights))
|
|
for j, addr := range params.addresses {
|
|
clients[j] = p.clientBuilder(addr)
|
|
if err := clients[j].dial(ctx); err != nil {
|
|
p.log(zap.WarnLevel, "failed to build client", zap.String("address", addr), zap.Error(err))
|
|
continue
|
|
}
|
|
|
|
var st session.Object
|
|
err := initSessionForDuration(ctx, &st, clients[j], p.rebalanceParams.sessionExpirationDuration, *p.key)
|
|
if err != nil {
|
|
clients[j].setUnhealthy()
|
|
p.log(zap.WarnLevel, "failed to create frostfs session token for client",
|
|
zap.String("address", addr), zap.Error(err))
|
|
continue
|
|
}
|
|
|
|
_ = p.cache.Put(formCacheKey(addr, p.key), st)
|
|
atLeastOneHealthy = true
|
|
}
|
|
source := rand.NewSource(time.Now().UnixNano())
|
|
sampl := newSampler(params.weights, source)
|
|
|
|
inner[i] = &innerPool{
|
|
sampler: sampl,
|
|
clients: clients,
|
|
}
|
|
}
|
|
|
|
if !atLeastOneHealthy {
|
|
return fmt.Errorf("at least one node must be healthy")
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
p.cancel = cancel
|
|
p.closedCh = make(chan struct{})
|
|
p.innerPools = inner
|
|
|
|
go p.startRebalance(ctx)
|
|
return nil
|
|
}
|
|
|
|
func (p *Pool) log(level zapcore.Level, msg string, fields ...zap.Field) {
|
|
if p.logger == nil {
|
|
return
|
|
}
|
|
|
|
p.logger.Log(level, msg, fields...)
|
|
}
|
|
|
|
func fillDefaultInitParams(params *InitParameters, cache *sessionCache) {
|
|
if params.sessionExpirationDuration == 0 {
|
|
params.sessionExpirationDuration = defaultSessionTokenExpirationDuration
|
|
}
|
|
|
|
if params.errorThreshold == 0 {
|
|
params.errorThreshold = defaultErrorThreshold
|
|
}
|
|
|
|
if params.clientRebalanceInterval <= 0 {
|
|
params.clientRebalanceInterval = defaultRebalanceInterval
|
|
}
|
|
|
|
if params.healthcheckTimeout <= 0 {
|
|
params.healthcheckTimeout = defaultHealthcheckTimeout
|
|
}
|
|
|
|
if params.nodeDialTimeout <= 0 {
|
|
params.nodeDialTimeout = defaultDialTimeout
|
|
}
|
|
|
|
if params.nodeStreamTimeout <= 0 {
|
|
params.nodeStreamTimeout = defaultStreamTimeout
|
|
}
|
|
|
|
if params.isMissingClientBuilder() {
|
|
params.setClientBuilder(func(addr string) client {
|
|
var prm wrapperPrm
|
|
prm.setAddress(addr)
|
|
prm.setKey(*params.key)
|
|
prm.setDialTimeout(params.nodeDialTimeout)
|
|
prm.setStreamTimeout(params.nodeStreamTimeout)
|
|
prm.setErrorThreshold(params.errorThreshold)
|
|
prm.setPoolRequestCallback(params.requestCallback)
|
|
prm.setGRPCDialOptions(params.dialOptions)
|
|
prm.setResponseInfoCallback(func(info sdkClient.ResponseMetaInfo) error {
|
|
cache.updateEpoch(info.Epoch())
|
|
return nil
|
|
})
|
|
return newWrapper(prm)
|
|
})
|
|
}
|
|
}
|
|
|
|
func adjustNodeParams(nodeParams []NodeParam) ([]*nodesParam, error) {
|
|
if len(nodeParams) == 0 {
|
|
return nil, errors.New("no FrostFS peers configured")
|
|
}
|
|
|
|
nodesParamsMap := make(map[int]*nodesParam)
|
|
for _, param := range nodeParams {
|
|
nodes, ok := nodesParamsMap[param.priority]
|
|
if !ok {
|
|
nodes = &nodesParam{priority: param.priority}
|
|
}
|
|
nodes.addresses = append(nodes.addresses, param.address)
|
|
nodes.weights = append(nodes.weights, param.weight)
|
|
nodesParamsMap[param.priority] = nodes
|
|
}
|
|
|
|
nodesParams := make([]*nodesParam, 0, len(nodesParamsMap))
|
|
for _, nodes := range nodesParamsMap {
|
|
nodes.weights = adjustWeights(nodes.weights)
|
|
nodesParams = append(nodesParams, nodes)
|
|
}
|
|
|
|
sort.Slice(nodesParams, func(i, j int) bool {
|
|
return nodesParams[i].priority < nodesParams[j].priority
|
|
})
|
|
|
|
return nodesParams, nil
|
|
}
|
|
|
|
// startRebalance runs loop to monitor connection healthy status.
|
|
func (p *Pool) startRebalance(ctx context.Context) {
|
|
ticker := time.NewTimer(p.rebalanceParams.clientRebalanceInterval)
|
|
buffers := make([][]float64, len(p.rebalanceParams.nodesParams))
|
|
for i, params := range p.rebalanceParams.nodesParams {
|
|
buffers[i] = make([]float64, len(params.weights))
|
|
}
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
close(p.closedCh)
|
|
return
|
|
case <-ticker.C:
|
|
p.updateNodesHealth(ctx, buffers)
|
|
ticker.Reset(p.rebalanceParams.clientRebalanceInterval)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (p *Pool) updateNodesHealth(ctx context.Context, buffers [][]float64) {
|
|
wg := sync.WaitGroup{}
|
|
for i, inner := range p.innerPools {
|
|
wg.Add(1)
|
|
|
|
bufferWeights := buffers[i]
|
|
go func(i int, innerPool *innerPool) {
|
|
defer wg.Done()
|
|
p.updateInnerNodesHealth(ctx, i, bufferWeights)
|
|
}(i, inner)
|
|
}
|
|
wg.Wait()
|
|
}
|
|
|
|
func (p *Pool) updateInnerNodesHealth(ctx context.Context, i int, bufferWeights []float64) {
|
|
if i > len(p.innerPools)-1 {
|
|
return
|
|
}
|
|
pool := p.innerPools[i]
|
|
options := p.rebalanceParams
|
|
|
|
healthyChanged := new(atomic.Bool)
|
|
wg := sync.WaitGroup{}
|
|
|
|
for j, cli := range pool.clients {
|
|
wg.Add(1)
|
|
go func(j int, cli client) {
|
|
defer wg.Done()
|
|
|
|
tctx, c := context.WithTimeout(ctx, options.nodeRequestTimeout)
|
|
defer c()
|
|
|
|
healthy, changed := cli.restartIfUnhealthy(tctx)
|
|
if healthy {
|
|
bufferWeights[j] = options.nodesParams[i].weights[j]
|
|
} else {
|
|
bufferWeights[j] = 0
|
|
p.cache.DeleteByPrefix(cli.address())
|
|
}
|
|
|
|
if changed {
|
|
p.log(zap.DebugLevel, "health has changed",
|
|
zap.String("address", cli.address()), zap.Bool("healthy", healthy))
|
|
healthyChanged.Store(true)
|
|
}
|
|
}(j, cli)
|
|
}
|
|
wg.Wait()
|
|
|
|
if healthyChanged.Load() {
|
|
probabilities := adjustWeights(bufferWeights)
|
|
source := rand.NewSource(time.Now().UnixNano())
|
|
pool.lock.Lock()
|
|
pool.sampler = newSampler(probabilities, source)
|
|
pool.lock.Unlock()
|
|
}
|
|
}
|
|
|
|
func adjustWeights(weights []float64) []float64 {
|
|
adjusted := make([]float64, len(weights))
|
|
sum := 0.0
|
|
for _, weight := range weights {
|
|
sum += weight
|
|
}
|
|
if sum > 0 {
|
|
for i, weight := range weights {
|
|
adjusted[i] = weight / sum
|
|
}
|
|
}
|
|
|
|
return adjusted
|
|
}
|
|
|
|
func (p *Pool) connection() (client, error) {
|
|
for _, inner := range p.innerPools {
|
|
cp, err := inner.connection()
|
|
if err == nil {
|
|
return cp, nil
|
|
}
|
|
}
|
|
|
|
return nil, errors.New("no healthy client")
|
|
}
|
|
|
|
func (p *innerPool) connection() (client, error) {
|
|
p.lock.RLock() // need lock because of using p.sampler
|
|
defer p.lock.RUnlock()
|
|
if len(p.clients) == 1 {
|
|
cp := p.clients[0]
|
|
if cp.isHealthy() {
|
|
return cp, nil
|
|
}
|
|
return nil, errors.New("no healthy client")
|
|
}
|
|
attempts := 3 * len(p.clients)
|
|
for k := 0; k < attempts; k++ {
|
|
i := p.sampler.Next()
|
|
if cp := p.clients[i]; cp.isHealthy() {
|
|
return cp, nil
|
|
}
|
|
}
|
|
|
|
return nil, errors.New("no healthy client")
|
|
}
|
|
|
|
func formCacheKey(address string, key *ecdsa.PrivateKey) string {
|
|
k := keys.PrivateKey{PrivateKey: *key}
|
|
return address + k.String()
|
|
}
|
|
|
|
func (p *Pool) checkSessionTokenErr(err error, address string) bool {
|
|
if err == nil {
|
|
return false
|
|
}
|
|
|
|
if sdkClient.IsErrSessionNotFound(err) || sdkClient.IsErrSessionExpired(err) {
|
|
p.cache.DeleteByPrefix(address)
|
|
return true
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
func initSessionForDuration(ctx context.Context, dst *session.Object, c client, dur uint64, ownerKey ecdsa.PrivateKey) error {
|
|
ni, err := c.networkInfo(ctx, prmNetworkInfo{})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
epoch := ni.CurrentEpoch()
|
|
|
|
var exp uint64
|
|
if math.MaxUint64-epoch < dur {
|
|
exp = math.MaxUint64
|
|
} else {
|
|
exp = epoch + dur
|
|
}
|
|
var prm prmCreateSession
|
|
prm.setExp(exp)
|
|
prm.useKey(ownerKey)
|
|
|
|
res, err := c.sessionCreate(ctx, prm)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
var id uuid.UUID
|
|
|
|
err = id.UnmarshalBinary(res.id)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid session token ID: %w", err)
|
|
}
|
|
|
|
var key frostfsecdsa.PublicKey
|
|
|
|
err = key.Decode(res.sessionKey)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid public session key: %w", err)
|
|
}
|
|
|
|
dst.SetID(id)
|
|
dst.SetAuthKey(&key)
|
|
dst.SetExp(exp)
|
|
|
|
return nil
|
|
}
|
|
|
|
type callContext struct {
|
|
client client
|
|
|
|
// client endpoint
|
|
endpoint string
|
|
|
|
// request signer
|
|
key *ecdsa.PrivateKey
|
|
|
|
// flag to open default session if session token is missing
|
|
sessionDefault bool
|
|
sessionTarget func(session.Object)
|
|
sessionVerb session.ObjectVerb
|
|
sessionCnr cid.ID
|
|
sessionObjSet bool
|
|
sessionObjs []oid.ID
|
|
}
|
|
|
|
func (p *Pool) initCallContext(ctx *callContext, cfg prmCommon, prmCtx prmContext) error {
|
|
cp, err := p.connection()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
ctx.key = cfg.key
|
|
if ctx.key == nil {
|
|
// use pool key if caller didn't specify its own
|
|
ctx.key = p.key
|
|
}
|
|
|
|
ctx.endpoint = cp.address()
|
|
ctx.client = cp
|
|
|
|
if ctx.sessionTarget != nil && cfg.stoken != nil {
|
|
ctx.sessionTarget(*cfg.stoken)
|
|
}
|
|
|
|
// note that we don't override session provided by the caller
|
|
ctx.sessionDefault = cfg.stoken == nil && prmCtx.defaultSession
|
|
if ctx.sessionDefault {
|
|
ctx.sessionVerb = prmCtx.verb
|
|
ctx.sessionCnr = prmCtx.cnr
|
|
ctx.sessionObjSet = prmCtx.objSet
|
|
ctx.sessionObjs = prmCtx.objs
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
// opens new session or uses cached one.
|
|
// Must be called only on initialized callContext with set sessionTarget.
|
|
func (p *Pool) openDefaultSession(ctx context.Context, cc *callContext) error {
|
|
cacheKey := formCacheKey(cc.endpoint, cc.key)
|
|
|
|
tok, ok := p.cache.Get(cacheKey)
|
|
if !ok {
|
|
// init new session
|
|
err := initSessionForDuration(ctx, &tok, cc.client, p.stokenDuration, *cc.key)
|
|
if err != nil {
|
|
return fmt.Errorf("session API client: %w", err)
|
|
}
|
|
|
|
// cache the opened session
|
|
p.cache.Put(cacheKey, tok)
|
|
}
|
|
|
|
tok.ForVerb(cc.sessionVerb)
|
|
tok.BindContainer(cc.sessionCnr)
|
|
|
|
if cc.sessionObjSet {
|
|
tok.LimitByObjects(cc.sessionObjs...)
|
|
}
|
|
|
|
// sign the token
|
|
if err := tok.Sign(*cc.key); err != nil {
|
|
return fmt.Errorf("sign token of the opened session: %w", err)
|
|
}
|
|
|
|
cc.sessionTarget(tok)
|
|
|
|
return nil
|
|
}
|
|
|
|
// opens default session (if sessionDefault is set), and calls f. If f returns
|
|
// session-related error then cached token is removed.
|
|
func (p *Pool) call(ctx context.Context, cc *callContext, f func() error) error {
|
|
var err error
|
|
|
|
if cc.sessionDefault {
|
|
err = p.openDefaultSession(ctx, cc)
|
|
if err != nil {
|
|
return fmt.Errorf("open default session: %w", err)
|
|
}
|
|
}
|
|
|
|
err = f()
|
|
_ = p.checkSessionTokenErr(err, cc.endpoint)
|
|
|
|
return err
|
|
}
|
|
|
|
// fillAppropriateKey use pool key if caller didn't specify its own.
|
|
func (p *Pool) fillAppropriateKey(prm *prmCommon) {
|
|
if prm.key == nil {
|
|
prm.key = p.key
|
|
}
|
|
}
|
|
|
|
// PutObject writes an object through a remote server using FrostFS API protocol.
|
|
//
|
|
// Main return value MUST NOT be processed on an erroneous return.
|
|
func (p *Pool) PutObject(ctx context.Context, prm PrmObjectPut) (oid.ID, error) {
|
|
cnr, _ := prm.hdr.ContainerID()
|
|
|
|
var prmCtx prmContext
|
|
prmCtx.useDefaultSession()
|
|
prmCtx.useVerb(session.VerbObjectPut)
|
|
prmCtx.useContainer(cnr)
|
|
|
|
p.fillAppropriateKey(&prm.prmCommon)
|
|
|
|
var ctxCall callContext
|
|
if err := p.initCallContext(&ctxCall, prm.prmCommon, prmCtx); err != nil {
|
|
return oid.ID{}, fmt.Errorf("init call context: %w", err)
|
|
}
|
|
|
|
if ctxCall.sessionDefault {
|
|
ctxCall.sessionTarget = prm.UseSession
|
|
if err := p.openDefaultSession(ctx, &ctxCall); err != nil {
|
|
return oid.ID{}, fmt.Errorf("open default session: %w", err)
|
|
}
|
|
}
|
|
|
|
id, err := ctxCall.client.objectPut(ctx, prm)
|
|
if err != nil {
|
|
// removes session token from cache in case of token error
|
|
p.checkSessionTokenErr(err, ctxCall.endpoint)
|
|
return id, fmt.Errorf("init writing on API client: %w", err)
|
|
}
|
|
|
|
return id, nil
|
|
}
|
|
|
|
// DeleteObject marks an object for deletion from the container using FrostFS API protocol.
|
|
// As a marker, a special unit called a tombstone is placed in the container.
|
|
// It confirms the user's intent to delete the object, and is itself a container object.
|
|
// Explicit deletion is done asynchronously, and is generally not guaranteed.
|
|
func (p *Pool) DeleteObject(ctx context.Context, prm PrmObjectDelete) error {
|
|
var prmCtx prmContext
|
|
prmCtx.useDefaultSession()
|
|
prmCtx.useVerb(session.VerbObjectDelete)
|
|
prmCtx.useAddress(prm.addr)
|
|
|
|
if prm.stoken == nil { // collect phy objects only if we are about to open default session
|
|
var tokens relations.Tokens
|
|
tokens.Bearer = prm.btoken
|
|
|
|
relatives, err := relations.ListAllRelations(ctx, p, prm.addr.Container(), prm.addr.Object(), tokens)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to collect relatives: %w", err)
|
|
}
|
|
|
|
if len(relatives) != 0 {
|
|
prmCtx.useContainer(prm.addr.Container())
|
|
prmCtx.useObjects(append(relatives, prm.addr.Object()))
|
|
}
|
|
}
|
|
|
|
p.fillAppropriateKey(&prm.prmCommon)
|
|
|
|
var cc callContext
|
|
cc.sessionTarget = prm.UseSession
|
|
|
|
err := p.initCallContext(&cc, prm.prmCommon, prmCtx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return p.call(ctx, &cc, func() error {
|
|
if err = cc.client.objectDelete(ctx, prm); err != nil {
|
|
return fmt.Errorf("remove object via client: %w", err)
|
|
}
|
|
|
|
return nil
|
|
})
|
|
}
|
|
|
|
type objectReadCloser struct {
|
|
reader *sdkClient.ObjectReader
|
|
elapsedTimeCallback func(time.Duration)
|
|
}
|
|
|
|
// Read implements io.Reader of the object payload.
|
|
func (x *objectReadCloser) Read(p []byte) (int, error) {
|
|
start := time.Now()
|
|
n, err := x.reader.Read(p)
|
|
x.elapsedTimeCallback(time.Since(start))
|
|
return n, err
|
|
}
|
|
|
|
// Close implements io.Closer of the object payload.
|
|
func (x *objectReadCloser) Close() error {
|
|
_, err := x.reader.Close()
|
|
return err
|
|
}
|
|
|
|
// ResGetObject is designed to provide object header nad read one object payload from FrostFS system.
|
|
type ResGetObject struct {
|
|
Header object.Object
|
|
|
|
Payload io.ReadCloser
|
|
}
|
|
|
|
// GetObject reads object header and initiates reading an object payload through a remote server using FrostFS API protocol.
|
|
//
|
|
// Main return value MUST NOT be processed on an erroneous return.
|
|
func (p *Pool) GetObject(ctx context.Context, prm PrmObjectGet) (ResGetObject, error) {
|
|
p.fillAppropriateKey(&prm.prmCommon)
|
|
|
|
var cc callContext
|
|
cc.sessionTarget = prm.UseSession
|
|
|
|
var res ResGetObject
|
|
|
|
err := p.initCallContext(&cc, prm.prmCommon, prmContext{})
|
|
if err != nil {
|
|
return res, err
|
|
}
|
|
|
|
return res, p.call(ctx, &cc, func() error {
|
|
res, err = cc.client.objectGet(ctx, prm)
|
|
return err
|
|
})
|
|
}
|
|
|
|
// HeadObject reads object header through a remote server using FrostFS API protocol.
|
|
//
|
|
// Main return value MUST NOT be processed on an erroneous return.
|
|
func (p *Pool) HeadObject(ctx context.Context, prm PrmObjectHead) (object.Object, error) {
|
|
p.fillAppropriateKey(&prm.prmCommon)
|
|
|
|
var cc callContext
|
|
cc.sessionTarget = prm.UseSession
|
|
|
|
var obj object.Object
|
|
|
|
err := p.initCallContext(&cc, prm.prmCommon, prmContext{})
|
|
if err != nil {
|
|
return obj, err
|
|
}
|
|
|
|
return obj, p.call(ctx, &cc, func() error {
|
|
obj, err = cc.client.objectHead(ctx, prm)
|
|
return err
|
|
})
|
|
}
|
|
|
|
// ResObjectRange is designed to read payload range of one object
|
|
// from FrostFS system.
|
|
//
|
|
// Must be initialized using Pool.ObjectRange, any other
|
|
// usage is unsafe.
|
|
type ResObjectRange struct {
|
|
payload *sdkClient.ObjectRangeReader
|
|
elapsedTimeCallback func(time.Duration)
|
|
}
|
|
|
|
// Read implements io.Reader of the object payload.
|
|
func (x *ResObjectRange) Read(p []byte) (int, error) {
|
|
start := time.Now()
|
|
n, err := x.payload.Read(p)
|
|
x.elapsedTimeCallback(time.Since(start))
|
|
return n, err
|
|
}
|
|
|
|
// Close ends reading the payload range and returns the result of the operation
|
|
// along with the final results. Must be called after using the ResObjectRange.
|
|
func (x *ResObjectRange) Close() error {
|
|
_, err := x.payload.Close()
|
|
return err
|
|
}
|
|
|
|
// ObjectRange initiates reading an object's payload range through a remote
|
|
// server using FrostFS API protocol.
|
|
//
|
|
// Main return value MUST NOT be processed on an erroneous return.
|
|
func (p *Pool) ObjectRange(ctx context.Context, prm PrmObjectRange) (ResObjectRange, error) {
|
|
p.fillAppropriateKey(&prm.prmCommon)
|
|
|
|
var cc callContext
|
|
cc.sessionTarget = prm.UseSession
|
|
|
|
var res ResObjectRange
|
|
|
|
err := p.initCallContext(&cc, prm.prmCommon, prmContext{})
|
|
if err != nil {
|
|
return res, err
|
|
}
|
|
|
|
return res, p.call(ctx, &cc, func() error {
|
|
res, err = cc.client.objectRange(ctx, prm)
|
|
return err
|
|
})
|
|
}
|
|
|
|
// ResObjectSearch is designed to read list of object identifiers from FrostFS system.
|
|
//
|
|
// Must be initialized using Pool.SearchObjects, any other usage is unsafe.
|
|
type ResObjectSearch struct {
|
|
r *sdkClient.ObjectListReader
|
|
}
|
|
|
|
// Read reads another list of the object identifiers.
|
|
func (x *ResObjectSearch) Read(buf []oid.ID) (int, error) {
|
|
n, ok := x.r.Read(buf)
|
|
if !ok {
|
|
_, err := x.r.Close()
|
|
if err == nil {
|
|
return n, io.EOF
|
|
}
|
|
|
|
return n, err
|
|
}
|
|
|
|
return n, nil
|
|
}
|
|
|
|
// Iterate iterates over the list of found object identifiers.
|
|
// f can return true to stop iteration earlier.
|
|
//
|
|
// Returns an error if object can't be read.
|
|
func (x *ResObjectSearch) Iterate(f func(oid.ID) bool) error {
|
|
return x.r.Iterate(f)
|
|
}
|
|
|
|
// Close ends reading list of the matched objects and returns the result of the operation
|
|
// along with the final results. Must be called after using the ResObjectSearch.
|
|
func (x *ResObjectSearch) Close() {
|
|
_, _ = x.r.Close()
|
|
}
|
|
|
|
// SearchObjects initiates object selection through a remote server using FrostFS API protocol.
|
|
//
|
|
// The call only opens the transmission channel, explicit fetching of matched objects
|
|
// is done using the ResObjectSearch. Resulting reader must be finally closed.
|
|
//
|
|
// Main return value MUST NOT be processed on an erroneous return.
|
|
func (p *Pool) SearchObjects(ctx context.Context, prm PrmObjectSearch) (ResObjectSearch, error) {
|
|
p.fillAppropriateKey(&prm.prmCommon)
|
|
|
|
var cc callContext
|
|
cc.sessionTarget = prm.UseSession
|
|
|
|
var res ResObjectSearch
|
|
|
|
err := p.initCallContext(&cc, prm.prmCommon, prmContext{})
|
|
if err != nil {
|
|
return res, err
|
|
}
|
|
|
|
return res, p.call(ctx, &cc, func() error {
|
|
res, err = cc.client.objectSearch(ctx, prm)
|
|
return err
|
|
})
|
|
}
|
|
|
|
// PutContainer sends request to save container in FrostFS and waits for the operation to complete.
|
|
//
|
|
// Waiting parameters can be specified using SetWaitParams. If not called, defaults are used:
|
|
//
|
|
// polling interval: 5s
|
|
// waiting timeout: 120s
|
|
//
|
|
// Success can be verified by reading by identifier (see GetContainer).
|
|
//
|
|
// Main return value MUST NOT be processed on an erroneous return.
|
|
func (p *Pool) PutContainer(ctx context.Context, prm PrmContainerPut) (cid.ID, error) {
|
|
cp, err := p.connection()
|
|
if err != nil {
|
|
return cid.ID{}, err
|
|
}
|
|
|
|
return cp.containerPut(ctx, prm)
|
|
}
|
|
|
|
// GetContainer reads FrostFS container by ID.
|
|
//
|
|
// Main return value MUST NOT be processed on an erroneous return.
|
|
func (p *Pool) GetContainer(ctx context.Context, prm PrmContainerGet) (container.Container, error) {
|
|
cp, err := p.connection()
|
|
if err != nil {
|
|
return container.Container{}, err
|
|
}
|
|
|
|
return cp.containerGet(ctx, prm)
|
|
}
|
|
|
|
// ListContainers requests identifiers of the account-owned containers.
|
|
func (p *Pool) ListContainers(ctx context.Context, prm PrmContainerList) ([]cid.ID, error) {
|
|
cp, err := p.connection()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return cp.containerList(ctx, prm)
|
|
}
|
|
|
|
// DeleteContainer sends request to remove the FrostFS container and waits for the operation to complete.
|
|
//
|
|
// Waiting parameters can be specified using SetWaitParams. If not called, defaults are used:
|
|
//
|
|
// polling interval: 5s
|
|
// waiting timeout: 120s
|
|
//
|
|
// Success can be verified by reading by identifier (see GetContainer).
|
|
func (p *Pool) DeleteContainer(ctx context.Context, prm PrmContainerDelete) error {
|
|
cp, err := p.connection()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return cp.containerDelete(ctx, prm)
|
|
}
|
|
|
|
// GetEACL reads eACL table of the FrostFS container.
|
|
//
|
|
// Main return value MUST NOT be processed on an erroneous return.
|
|
func (p *Pool) GetEACL(ctx context.Context, prm PrmContainerEACL) (eacl.Table, error) {
|
|
cp, err := p.connection()
|
|
if err != nil {
|
|
return eacl.Table{}, err
|
|
}
|
|
|
|
return cp.containerEACL(ctx, prm)
|
|
}
|
|
|
|
// SetEACL sends request to update eACL table of the FrostFS container and waits for the operation to complete.
|
|
//
|
|
// Waiting parameters can be specified using SetWaitParams. If not called, defaults are used:
|
|
//
|
|
// polling interval: 5s
|
|
// waiting timeout: 120s
|
|
//
|
|
// Success can be verified by reading by identifier (see GetEACL).
|
|
func (p *Pool) SetEACL(ctx context.Context, prm PrmContainerSetEACL) error {
|
|
cp, err := p.connection()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return cp.containerSetEACL(ctx, prm)
|
|
}
|
|
|
|
// Balance requests current balance of the FrostFS account.
|
|
//
|
|
// Main return value MUST NOT be processed on an erroneous return.
|
|
func (p *Pool) Balance(ctx context.Context, prm PrmBalanceGet) (accounting.Decimal, error) {
|
|
cp, err := p.connection()
|
|
if err != nil {
|
|
return accounting.Decimal{}, err
|
|
}
|
|
|
|
return cp.balanceGet(ctx, prm)
|
|
}
|
|
|
|
// Statistic returns connection statistics.
|
|
func (p Pool) Statistic() Statistic {
|
|
stat := Statistic{}
|
|
for _, inner := range p.innerPools {
|
|
inner.lock.RLock()
|
|
for _, cl := range inner.clients {
|
|
node := NodeStatistic{
|
|
address: cl.address(),
|
|
methods: cl.methodsStatus(),
|
|
overallErrors: cl.overallErrorRate(),
|
|
currentErrors: cl.currentErrorRate(),
|
|
}
|
|
stat.nodes = append(stat.nodes, node)
|
|
stat.overallErrors += node.overallErrors
|
|
}
|
|
inner.lock.RUnlock()
|
|
}
|
|
|
|
return stat
|
|
}
|
|
|
|
// waitForContainerPresence waits until the container is found on the FrostFS network.
|
|
func waitForContainerPresence(ctx context.Context, cli client, cnrID cid.ID, waitParams *WaitParams) error {
|
|
var prm PrmContainerGet
|
|
prm.SetContainerID(cnrID)
|
|
|
|
return waitFor(ctx, waitParams, func(ctx context.Context) bool {
|
|
_, err := cli.containerGet(ctx, prm)
|
|
return err == nil
|
|
})
|
|
}
|
|
|
|
// waitForEACLPresence waits until the container eacl is applied on the FrostFS network.
|
|
func waitForEACLPresence(ctx context.Context, cli client, cnrID *cid.ID, table *eacl.Table, waitParams *WaitParams) error {
|
|
var prm PrmContainerEACL
|
|
if cnrID != nil {
|
|
prm.SetContainerID(*cnrID)
|
|
}
|
|
|
|
return waitFor(ctx, waitParams, func(ctx context.Context) bool {
|
|
eaclTable, err := cli.containerEACL(ctx, prm)
|
|
if err == nil {
|
|
return eacl.EqualTables(*table, eaclTable)
|
|
}
|
|
return false
|
|
})
|
|
}
|
|
|
|
// waitForContainerRemoved waits until the container is removed from the FrostFS network.
|
|
func waitForContainerRemoved(ctx context.Context, cli client, cnrID *cid.ID, waitParams *WaitParams) error {
|
|
var prm PrmContainerGet
|
|
if cnrID != nil {
|
|
prm.SetContainerID(*cnrID)
|
|
}
|
|
|
|
return waitFor(ctx, waitParams, func(ctx context.Context) bool {
|
|
_, err := cli.containerGet(ctx, prm)
|
|
return sdkClient.IsErrContainerNotFound(err)
|
|
})
|
|
}
|
|
|
|
// waitFor await that given condition will be met in waitParams time.
|
|
func waitFor(ctx context.Context, params *WaitParams, condition func(context.Context) bool) error {
|
|
wctx, cancel := context.WithTimeout(ctx, params.timeout)
|
|
defer cancel()
|
|
ticker := time.NewTimer(params.pollInterval)
|
|
defer ticker.Stop()
|
|
wdone := wctx.Done()
|
|
done := ctx.Done()
|
|
for {
|
|
select {
|
|
case <-done:
|
|
return ctx.Err()
|
|
case <-wdone:
|
|
return wctx.Err()
|
|
case <-ticker.C:
|
|
if condition(ctx) {
|
|
return nil
|
|
}
|
|
ticker.Reset(params.pollInterval)
|
|
}
|
|
}
|
|
}
|
|
|
|
// NetworkInfo requests information about the FrostFS network of which the remote server is a part.
|
|
//
|
|
// Main return value MUST NOT be processed on an erroneous return.
|
|
func (p *Pool) NetworkInfo(ctx context.Context) (netmap.NetworkInfo, error) {
|
|
cp, err := p.connection()
|
|
if err != nil {
|
|
return netmap.NetworkInfo{}, err
|
|
}
|
|
|
|
return cp.networkInfo(ctx, prmNetworkInfo{})
|
|
}
|
|
|
|
// Close closes the Pool and releases all the associated resources.
|
|
func (p *Pool) Close() {
|
|
p.cancel()
|
|
<-p.closedCh
|
|
}
|
|
|
|
// SyncContainerWithNetwork applies network configuration received via
|
|
// the Pool to the container. Changes the container if it does not satisfy
|
|
// network configuration.
|
|
//
|
|
// Pool and container MUST not be nil.
|
|
//
|
|
// Returns any error that does not allow reading configuration
|
|
// from the network.
|
|
func SyncContainerWithNetwork(ctx context.Context, cnr *container.Container, p *Pool) error {
|
|
ni, err := p.NetworkInfo(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("network info: %w", err)
|
|
}
|
|
|
|
container.ApplyNetworkConfig(cnr, ni)
|
|
|
|
return nil
|
|
}
|
|
|
|
// GetSplitInfo implements relations.Relations.
|
|
func (p *Pool) GetSplitInfo(ctx context.Context, cnrID cid.ID, objID oid.ID, tokens relations.Tokens) (*object.SplitInfo, error) {
|
|
var addr oid.Address
|
|
addr.SetContainer(cnrID)
|
|
addr.SetObject(objID)
|
|
|
|
var prm PrmObjectHead
|
|
prm.SetAddress(addr)
|
|
if tokens.Bearer != nil {
|
|
prm.UseBearer(*tokens.Bearer)
|
|
}
|
|
if tokens.Session != nil {
|
|
prm.UseSession(*tokens.Session)
|
|
}
|
|
prm.MarkRaw()
|
|
|
|
_, err := p.HeadObject(ctx, prm)
|
|
|
|
var errSplit *object.SplitInfoError
|
|
|
|
switch {
|
|
case errors.As(err, &errSplit):
|
|
return errSplit.SplitInfo(), nil
|
|
case err == nil:
|
|
return nil, relations.ErrNoSplitInfo
|
|
default:
|
|
return nil, fmt.Errorf("failed to get raw object header: %w", err)
|
|
}
|
|
}
|
|
|
|
// ListChildrenByLinker implements relations.Relations.
|
|
func (p *Pool) ListChildrenByLinker(ctx context.Context, cnrID cid.ID, objID oid.ID, tokens relations.Tokens) ([]oid.ID, error) {
|
|
var addr oid.Address
|
|
addr.SetContainer(cnrID)
|
|
addr.SetObject(objID)
|
|
|
|
var prm PrmObjectHead
|
|
prm.SetAddress(addr)
|
|
if tokens.Bearer != nil {
|
|
prm.UseBearer(*tokens.Bearer)
|
|
}
|
|
if tokens.Session != nil {
|
|
prm.UseSession(*tokens.Session)
|
|
}
|
|
|
|
res, err := p.HeadObject(ctx, prm)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to get linking object's header: %w", err)
|
|
}
|
|
|
|
return res.Children(), nil
|
|
}
|
|
|
|
// GetLeftSibling implements relations.Relations.
|
|
func (p *Pool) GetLeftSibling(ctx context.Context, cnrID cid.ID, objID oid.ID, tokens relations.Tokens) (oid.ID, error) {
|
|
var addr oid.Address
|
|
addr.SetContainer(cnrID)
|
|
addr.SetObject(objID)
|
|
|
|
var prm PrmObjectHead
|
|
prm.SetAddress(addr)
|
|
if tokens.Bearer != nil {
|
|
prm.UseBearer(*tokens.Bearer)
|
|
}
|
|
if tokens.Session != nil {
|
|
prm.UseSession(*tokens.Session)
|
|
}
|
|
|
|
res, err := p.HeadObject(ctx, prm)
|
|
if err != nil {
|
|
return oid.ID{}, fmt.Errorf("failed to read split chain member's header: %w", err)
|
|
}
|
|
|
|
idMember, ok := res.PreviousID()
|
|
if !ok {
|
|
return oid.ID{}, relations.ErrNoLeftSibling
|
|
}
|
|
return idMember, nil
|
|
}
|
|
|
|
// FindSiblingBySplitID implements relations.Relations.
|
|
func (p *Pool) FindSiblingBySplitID(ctx context.Context, cnrID cid.ID, splitID *object.SplitID, tokens relations.Tokens) ([]oid.ID, error) {
|
|
var query object.SearchFilters
|
|
query.AddSplitIDFilter(object.MatchStringEqual, splitID)
|
|
|
|
var prm PrmObjectSearch
|
|
prm.SetContainerID(cnrID)
|
|
prm.SetFilters(query)
|
|
if tokens.Bearer != nil {
|
|
prm.UseBearer(*tokens.Bearer)
|
|
}
|
|
if tokens.Session != nil {
|
|
prm.UseSession(*tokens.Session)
|
|
}
|
|
|
|
res, err := p.SearchObjects(ctx, prm)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to search objects by split ID: %w", err)
|
|
}
|
|
|
|
var members []oid.ID
|
|
err = res.Iterate(func(id oid.ID) bool {
|
|
members = append(members, id)
|
|
return false
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to iterate found objects: %w", err)
|
|
}
|
|
|
|
return members, nil
|
|
}
|
|
|
|
// FindSiblingByParentID implements relations.Relations.
|
|
func (p *Pool) FindSiblingByParentID(ctx context.Context, cnrID cid.ID, objID oid.ID, tokens relations.Tokens) ([]oid.ID, error) {
|
|
var query object.SearchFilters
|
|
query.AddParentIDFilter(object.MatchStringEqual, objID)
|
|
|
|
var prm PrmObjectSearch
|
|
prm.SetContainerID(cnrID)
|
|
prm.SetFilters(query)
|
|
if tokens.Bearer != nil {
|
|
prm.UseBearer(*tokens.Bearer)
|
|
}
|
|
if tokens.Session != nil {
|
|
prm.UseSession(*tokens.Session)
|
|
}
|
|
|
|
resSearch, err := p.SearchObjects(ctx, prm)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to find object children: %w", err)
|
|
}
|
|
|
|
var res []oid.ID
|
|
err = resSearch.Iterate(func(id oid.ID) bool {
|
|
res = append(res, id)
|
|
return false
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to iterate found objects: %w", err)
|
|
}
|
|
|
|
return res, nil
|
|
}
|