frostfs-sdk-go/pool/pool.go

package pool

import (
	"context"
	"crypto/ecdsa"
	"errors"
	"fmt"
	"math"
	"math/rand"
	"strings"
	"sync"
	"time"

	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
	"github.com/nspcc-dev/neofs-sdk-go/client"
	"github.com/nspcc-dev/neofs-sdk-go/container"
	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
	"github.com/nspcc-dev/neofs-sdk-go/eacl"
	"github.com/nspcc-dev/neofs-sdk-go/object"
	"github.com/nspcc-dev/neofs-sdk-go/owner"
	"github.com/nspcc-dev/neofs-sdk-go/session"
	"go.uber.org/zap"
)

// Client is a wrapper for client.Client to generate mock.
type Client interface {
	client.Client
}

// BuilderOptions contains options used to build connection pool.
type BuilderOptions struct {
	Key                     *ecdsa.PrivateKey
	Logger                  *zap.Logger
	NodeConnectionTimeout   time.Duration
	NodeRequestTimeout      time.Duration
	ClientRebalanceInterval time.Duration
	SessionExpirationEpoch  uint64
	weights                 []float64
	addresses               []string
	clientBuilder           func(opts ...client.Option) (client.Client, error)
}

// Builder is an interim structure used to collect node addresses/weights and
// build connection pool subsequently.
type Builder struct {
	addresses []string
	weights   []float64
}

// ContainerPollingParams contains parameters used in polling is a container created or not.
type ContainerPollingParams struct {
	CreationTimeout time.Duration
	PollInterval    time.Duration
}

// DefaultPollingParams creates ContainerPollingParams with default values.
func DefaultPollingParams() *ContainerPollingParams {
	return &ContainerPollingParams{
		CreationTimeout: 120 * time.Second,
		PollInterval:    5 * time.Second,
	}
}

// AddNode adds address/weight pair to node PoolBuilder list.
func (pb *Builder) AddNode(address string, weight float64) *Builder {
	pb.addresses = append(pb.addresses, address)
	pb.weights = append(pb.weights, weight)
	return pb
}

// Build creates new pool based on current PoolBuilder state and options.
func (pb *Builder) Build(ctx context.Context, options *BuilderOptions) (Pool, error) {
	if len(pb.addresses) == 0 {
		return nil, errors.New("no NeoFS peers configured")
	}

	options.weights = adjustWeights(pb.weights)
	options.addresses = pb.addresses

	if options.clientBuilder == nil {
		options.clientBuilder = client.New
	}

	return newPool(ctx, options)
}

// Pool is an interface providing connection artifacts on request.
type Pool interface {
	client.Object
	client.Container
	Connection() (client.Client, *session.Token, error)
	OwnerID() *owner.ID
	WaitForContainerPresence(context.Context, *cid.ID, *ContainerPollingParams) error
	Close()

	ClientParam
}

// ClientParam is analogue client.Object, client.Container but uses session token cache.
type ClientParam interface {
	PutObjectParam(ctx context.Context, params *client.PutObjectParams, callParam *CallParam) (*object.ID, error)
	DeleteObjectParam(ctx context.Context, params *client.DeleteObjectParams, callParam *CallParam) error
	GetObjectParam(ctx context.Context, params *client.GetObjectParams, callParam *CallParam) (*object.Object, error)
	GetObjectHeaderParam(ctx context.Context, params *client.ObjectHeaderParams, callParam *CallParam) (*object.Object, error)
	ObjectPayloadRangeDataParam(ctx context.Context, params *client.RangeDataParams, callParam *CallParam) ([]byte, error)
	ObjectPayloadRangeSHA256Param(ctx context.Context, params *client.RangeChecksumParams, callParam *CallParam) ([][32]byte, error)
	ObjectPayloadRangeTZParam(ctx context.Context, params *client.RangeChecksumParams, callParam *CallParam) ([][64]byte, error)
	SearchObjectParam(ctx context.Context, params *client.SearchObjectParams, callParam *CallParam) ([]*object.ID, error)
	PutContainerParam(ctx context.Context, cnr *container.Container, callParam *CallParam) (*cid.ID, error)
	GetContainerParam(ctx context.Context, cid *cid.ID, callParam *CallParam) (*container.Container, error)
	ListContainersParam(ctx context.Context, ownerID *owner.ID, callParam *CallParam) ([]*cid.ID, error)
	DeleteContainerParam(ctx context.Context, cid *cid.ID, callParam *CallParam) error
	GetEACLParam(ctx context.Context, cid *cid.ID, callParam *CallParam) (*client.EACLWithSignature, error)
	SetEACLParam(ctx context.Context, table *eacl.Table, callParam *CallParam) error
	AnnounceContainerUsedSpaceParam(ctx context.Context, announce []container.UsedSpaceAnnouncement, callParam *CallParam) error
}

type clientPack struct {
	client  client.Client
	healthy bool
	address string
}

type CallParam struct {
	isRetry bool
	Key     *ecdsa.PrivateKey

	Options []client.CallOption
}

var _ Pool = (*pool)(nil)

type pool struct {
	lock        sync.RWMutex
	sampler     *Sampler
	key         *ecdsa.PrivateKey
	owner       *owner.ID
	clientPacks []*clientPack
	cancel      context.CancelFunc
	closedCh    chan struct{}
	cache       *SessionCache
}

func newPool(ctx context.Context, options *BuilderOptions) (Pool, error) {
	cache := NewCache()

	clientPacks := make([]*clientPack, len(options.weights))
	var atLeastOneHealthy bool
	for i, address := range options.addresses {
		c, err := options.clientBuilder(client.WithDefaultPrivateKey(options.Key),
			client.WithURIAddress(address, nil),
			client.WithDialTimeout(options.NodeConnectionTimeout))
		if err != nil {
			return nil, err
		}
		var healthy bool
		st, err := c.CreateSession(ctx, options.SessionExpirationEpoch)
		if err != nil && options.Logger != nil {
			options.Logger.Warn("failed to create neofs session token for client",
				zap.String("address", address),
				zap.Error(err))
		} else if err == nil {
			healthy, atLeastOneHealthy = true, true
			_ = cache.Put(formCacheKey(address, options.Key), st)
		}
		clientPacks[i] = &clientPack{client: c, healthy: healthy, address: address}
	}

	if !atLeastOneHealthy {
		return nil, fmt.Errorf("at least one node must be healthy")
	}

	source := rand.NewSource(time.Now().UnixNano())
	sampler := NewSampler(options.weights, source)
	wallet, err := owner.NEO3WalletFromPublicKey(&options.Key.PublicKey)
	if err != nil {
		return nil, err
	}
	ownerID := owner.NewIDFromNeo3Wallet(wallet)

	ctx, cancel := context.WithCancel(ctx)
	pool := &pool{
		sampler:     sampler,
		key:         options.Key,
		owner:       ownerID,
		clientPacks: clientPacks,
		cancel:      cancel,
		closedCh:    make(chan struct{}),
		cache:       cache,
	}
	go startRebalance(ctx, pool, options)
	return pool, nil
}

func startRebalance(ctx context.Context, p *pool, options *BuilderOptions) {
	ticker := time.NewTimer(options.ClientRebalanceInterval)
	buffer := make([]float64, len(options.weights))

	for {
		select {
		case <-ctx.Done():
			close(p.closedCh)
			return
		case <-ticker.C:
			updateNodesHealth(ctx, p, options, buffer)
			ticker.Reset(options.ClientRebalanceInterval)
		}
	}
}

func updateNodesHealth(ctx context.Context, p *pool, options *BuilderOptions, bufferWeights []float64) {
	if len(bufferWeights) != len(p.clientPacks) {
		bufferWeights = make([]float64, len(p.clientPacks))
	}
	healthyChanged := false
	wg := sync.WaitGroup{}
	for i, cPack := range p.clientPacks {
		wg.Add(1)

		go func(i int, client client.Client) {
			defer wg.Done()
			ok := true
			tctx, c := context.WithTimeout(ctx, options.NodeRequestTimeout)
			defer c()
			if _, err := client.EndpointInfo(tctx); err != nil {
				ok = false
				bufferWeights[i] = 0
			}
			p.lock.RLock()
			cp := *p.clientPacks[i]
			p.lock.RUnlock()

			if ok {
				bufferWeights[i] = options.weights[i]
				if !cp.healthy {
					if tkn, err := client.CreateSession(ctx, options.SessionExpirationEpoch); err != nil {
						ok = false
						bufferWeights[i] = 0
					} else {
						_ = p.cache.Put(formCacheKey(cp.address, p.key), tkn)
					}
				}
			} else {
				p.cache.DeleteByPrefix(cp.address)
			}

			p.lock.Lock()
			if p.clientPacks[i].healthy != ok {
				p.clientPacks[i].healthy = ok
				healthyChanged = true
			}
			p.lock.Unlock()
		}(i, cPack.client)
	}
	wg.Wait()

	if healthyChanged {
		probabilities := adjustWeights(bufferWeights)
		source := rand.NewSource(time.Now().UnixNano())
		p.lock.Lock()
		p.sampler = NewSampler(probabilities, source)
		p.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.Client, *session.Token, error) {
	cp, err := p.connection()
	if err != nil {
		return nil, nil, err
	}

	token := p.cache.Get(formCacheKey(cp.address, p.key))
	return cp.client, token, nil
}

func (p *pool) connection() (*clientPack, error) {
	p.lock.RLock()
	defer p.lock.RUnlock()
	if len(p.clientPacks) == 1 {
		cp := p.clientPacks[0]
		if cp.healthy {
			return cp, nil
		}
		return nil, errors.New("no healthy client")
	}
	attempts := 3 * len(p.clientPacks)
	for k := 0; k < attempts; k++ {
		i := p.sampler.Next()
		if cp := p.clientPacks[i]; cp.healthy {
			return cp, nil
		}
	}
	return nil, errors.New("no healthy client")
}

func (p *pool) OwnerID() *owner.ID {
	return p.owner
}

func (p *pool) conn(option []client.CallOption) (client.Client, []client.CallOption, error) {
	conn, token, err := p.Connection()
	if err != nil {
		return nil, nil, err
	}
	return conn, append([]client.CallOption{client.WithSession(token)}, option...), nil
}

func formCacheKey(address string, key *ecdsa.PrivateKey) string {
	k := keys.PrivateKey{PrivateKey: *key}
	return address + k.String()
}

func (p *pool) connParam(ctx context.Context, param *CallParam) (*clientPack, []client.CallOption, error) {
	cp, err := p.connection()
	if err != nil {
		return nil, nil, err
	}

	key := p.key
	if param.Key != nil {
		key = param.Key
	}

	param.Options = append(param.Options, client.WithKey(key))
	cacheKey := formCacheKey(cp.address, key)
	token := p.cache.Get(cacheKey)
	if token == nil {
		token, err = cp.client.CreateSession(ctx, math.MaxUint32, param.Options...)
		if err != nil {
			return nil, nil, err
		}
		_ = p.cache.Put(cacheKey, token)
	}

	return cp, append([]client.CallOption{client.WithSession(token)}, param.Options...), nil
}

func (p *pool) PutObject(ctx context.Context, params *client.PutObjectParams, option ...client.CallOption) (*object.ID, error) {
	conn, options, err := p.conn(option)
	if err != nil {
		return nil, err
	}
	return conn.PutObject(ctx, params, options...)
}

func (p *pool) DeleteObject(ctx context.Context, params *client.DeleteObjectParams, option ...client.CallOption) error {
	conn, options, err := p.conn(option)
	if err != nil {
		return err
	}
	return conn.DeleteObject(ctx, params, options...)
}

func (p *pool) GetObject(ctx context.Context, params *client.GetObjectParams, option ...client.CallOption) (*object.Object, error) {
	conn, options, err := p.conn(option)
	if err != nil {
		return nil, err
	}
	return conn.GetObject(ctx, params, options...)
}

func (p *pool) GetObjectHeader(ctx context.Context, params *client.ObjectHeaderParams, option ...client.CallOption) (*object.Object, error) {
	conn, options, err := p.conn(option)
	if err != nil {
		return nil, err
	}
	return conn.GetObjectHeader(ctx, params, options...)
}

func (p *pool) ObjectPayloadRangeData(ctx context.Context, params *client.RangeDataParams, option ...client.CallOption) ([]byte, error) {
	conn, options, err := p.conn(option)
	if err != nil {
		return nil, err
	}
	return conn.ObjectPayloadRangeData(ctx, params, options...)
}

func (p *pool) ObjectPayloadRangeSHA256(ctx context.Context, params *client.RangeChecksumParams, option ...client.CallOption) ([][32]byte, error) {
	conn, options, err := p.conn(option)
	if err != nil {
		return nil, err
	}
	return conn.ObjectPayloadRangeSHA256(ctx, params, options...)
}

func (p *pool) ObjectPayloadRangeTZ(ctx context.Context, params *client.RangeChecksumParams, option ...client.CallOption) ([][64]byte, error) {
	conn, options, err := p.conn(option)
	if err != nil {
		return nil, err
	}
	return conn.ObjectPayloadRangeTZ(ctx, params, options...)
}

func (p *pool) SearchObject(ctx context.Context, params *client.SearchObjectParams, option ...client.CallOption) ([]*object.ID, error) {
	conn, options, err := p.conn(option)
	if err != nil {
		return nil, err
	}
	return conn.SearchObject(ctx, params, options...)
}

func (p *pool) PutContainer(ctx context.Context, cnr *container.Container, option ...client.CallOption) (*cid.ID, error) {
	conn, options, err := p.conn(option)
	if err != nil {
		return nil, err
	}
	return conn.PutContainer(ctx, cnr, options...)
}

func (p *pool) GetContainer(ctx context.Context, cid *cid.ID, option ...client.CallOption) (*container.Container, error) {
	conn, options, err := p.conn(option)
	if err != nil {
		return nil, err
	}
	return conn.GetContainer(ctx, cid, options...)
}

func (p *pool) ListContainers(ctx context.Context, ownerID *owner.ID, option ...client.CallOption) ([]*cid.ID, error) {
	conn, options, err := p.conn(option)
	if err != nil {
		return nil, err
	}
	return conn.ListContainers(ctx, ownerID, options...)
}

func (p *pool) DeleteContainer(ctx context.Context, cid *cid.ID, option ...client.CallOption) error {
	conn, options, err := p.conn(option)
	if err != nil {
		return err
	}
	return conn.DeleteContainer(ctx, cid, options...)
}

func (p *pool) GetEACL(ctx context.Context, cid *cid.ID, option ...client.CallOption) (*client.EACLWithSignature, error) {
	conn, options, err := p.conn(option)
	if err != nil {
		return nil, err
	}
	return conn.GetEACL(ctx, cid, options...)
}

func (p *pool) SetEACL(ctx context.Context, table *eacl.Table, option ...client.CallOption) error {
	conn, options, err := p.conn(option)
	if err != nil {
		return err
	}
	return conn.SetEACL(ctx, table, options...)
}

func (p *pool) AnnounceContainerUsedSpace(ctx context.Context, announce []container.UsedSpaceAnnouncement, option ...client.CallOption) error {
	conn, options, err := p.conn(option)
	if err != nil {
		return err
	}
	return conn.AnnounceContainerUsedSpace(ctx, announce, options...)
}

func (p *pool) checkSessionTokenErr(err error, address string) bool {
	if err == nil {
		return false
	}

	if strings.Contains(err.Error(), "session token does not exist") {
		p.cache.DeleteByPrefix(address)
		return true
	}

	return false
}

func (p *pool) PutObjectParam(ctx context.Context, params *client.PutObjectParams, callParam *CallParam) (*object.ID, error) {
	cp, options, err := p.connParam(ctx, callParam)
	if err != nil {
		return nil, err
	}
	res, err := cp.client.PutObject(ctx, params, options...)
	if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry {
		callParam.isRetry = true
		return p.PutObjectParam(ctx, params, callParam)
	}
	return res, err
}

func (p *pool) DeleteObjectParam(ctx context.Context, params *client.DeleteObjectParams, callParam *CallParam) error {
	cp, options, err := p.connParam(ctx, callParam)
	if err != nil {
		return err
	}
	err = cp.client.DeleteObject(ctx, params, options...)
	if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry {
		callParam.isRetry = true
		return p.DeleteObjectParam(ctx, params, callParam)
	}
	return err
}

func (p *pool) GetObjectParam(ctx context.Context, params *client.GetObjectParams, callParam *CallParam) (*object.Object, error) {
	cp, options, err := p.connParam(ctx, callParam)
	if err != nil {
		return nil, err
	}
	res, err := cp.client.GetObject(ctx, params, options...)
	if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry {
		callParam.isRetry = true
		return p.GetObjectParam(ctx, params, callParam)
	}
	return res, err
}

func (p *pool) GetObjectHeaderParam(ctx context.Context, params *client.ObjectHeaderParams, callParam *CallParam) (*object.Object, error) {
	cp, options, err := p.connParam(ctx, callParam)
	if err != nil {
		return nil, err
	}
	res, err := cp.client.GetObjectHeader(ctx, params, options...)
	if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry {
		callParam.isRetry = true
		return p.GetObjectHeaderParam(ctx, params, callParam)
	}
	return res, err
}

func (p *pool) ObjectPayloadRangeDataParam(ctx context.Context, params *client.RangeDataParams, callParam *CallParam) ([]byte, error) {
	cp, options, err := p.connParam(ctx, callParam)
	if err != nil {
		return nil, err
	}
	res, err := cp.client.ObjectPayloadRangeData(ctx, params, options...)
	if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry {
		callParam.isRetry = true
		return p.ObjectPayloadRangeDataParam(ctx, params, callParam)
	}
	return res, err
}

func (p *pool) ObjectPayloadRangeSHA256Param(ctx context.Context, params *client.RangeChecksumParams, callParam *CallParam) ([][32]byte, error) {
	cp, options, err := p.connParam(ctx, callParam)
	if err != nil {
		return nil, err
	}
	res, err := cp.client.ObjectPayloadRangeSHA256(ctx, params, options...)
	if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry {
		callParam.isRetry = true
		return p.ObjectPayloadRangeSHA256Param(ctx, params, callParam)
	}
	return res, err
}

func (p *pool) ObjectPayloadRangeTZParam(ctx context.Context, params *client.RangeChecksumParams, callParam *CallParam) ([][64]byte, error) {
	cp, options, err := p.connParam(ctx, callParam)
	if err != nil {
		return nil, err
	}
	res, err := cp.client.ObjectPayloadRangeTZ(ctx, params, options...)
	if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry {
		callParam.isRetry = true
		return p.ObjectPayloadRangeTZParam(ctx, params, callParam)
	}
	return res, err
}

func (p *pool) SearchObjectParam(ctx context.Context, params *client.SearchObjectParams, callParam *CallParam) ([]*object.ID, error) {
	cp, options, err := p.connParam(ctx, callParam)
	if err != nil {
		return nil, err
	}
	res, err := cp.client.SearchObject(ctx, params, options...)
	if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry {
		callParam.isRetry = true
		return p.SearchObjectParam(ctx, params, callParam)
	}
	return res, err
}

func (p *pool) PutContainerParam(ctx context.Context, cnr *container.Container, callParam *CallParam) (*cid.ID, error) {
	cp, options, err := p.connParam(ctx, callParam)
	if err != nil {
		return nil, err
	}
	res, err := cp.client.PutContainer(ctx, cnr, options...)
	if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry {
		callParam.isRetry = true
		return p.PutContainerParam(ctx, cnr, callParam)
	}
	return res, err
}

func (p *pool) GetContainerParam(ctx context.Context, cid *cid.ID, callParam *CallParam) (*container.Container, error) {
	cp, options, err := p.connParam(ctx, callParam)
	if err != nil {
		return nil, err
	}
	res, err := cp.client.GetContainer(ctx, cid, options...)
	if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry {
		callParam.isRetry = true
		return p.GetContainerParam(ctx, cid, callParam)
	}
	return res, err
}

func (p *pool) ListContainersParam(ctx context.Context, ownerID *owner.ID, callParam *CallParam) ([]*cid.ID, error) {
	cp, options, err := p.connParam(ctx, callParam)
	if err != nil {
		return nil, err
	}
	res, err := cp.client.ListContainers(ctx, ownerID, options...)
	if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry {
		callParam.isRetry = true
		return p.ListContainersParam(ctx, ownerID, callParam)
	}
	return res, err
}

func (p *pool) DeleteContainerParam(ctx context.Context, cid *cid.ID, callParam *CallParam) error {
	cp, options, err := p.connParam(ctx, callParam)
	if err != nil {
		return err
	}
	err = cp.client.DeleteContainer(ctx, cid, options...)
	if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry {
		callParam.isRetry = true
		return p.DeleteContainerParam(ctx, cid, callParam)
	}
	return err
}

func (p *pool) GetEACLParam(ctx context.Context, cid *cid.ID, callParam *CallParam) (*client.EACLWithSignature, error) {
	cp, options, err := p.connParam(ctx, callParam)
	if err != nil {
		return nil, err
	}
	res, err := cp.client.GetEACL(ctx, cid, options...)
	if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry {
		callParam.isRetry = true
		return p.GetEACLParam(ctx, cid, callParam)
	}
	return res, err
}

func (p *pool) SetEACLParam(ctx context.Context, table *eacl.Table, callParam *CallParam) error {
	cp, options, err := p.connParam(ctx, callParam)
	if err != nil {
		return err
	}
	err = cp.client.SetEACL(ctx, table, options...)
	if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry {
		callParam.isRetry = true
		return p.SetEACLParam(ctx, table, callParam)
	}
	return err
}

func (p *pool) AnnounceContainerUsedSpaceParam(ctx context.Context, announce []container.UsedSpaceAnnouncement, callParam *CallParam) error {
	cp, options, err := p.connParam(ctx, callParam)
	if err != nil {
		return err
	}
	err = cp.client.AnnounceContainerUsedSpace(ctx, announce, options...)
	if p.checkSessionTokenErr(err, cp.address) && !callParam.isRetry {
		callParam.isRetry = true
		return p.AnnounceContainerUsedSpaceParam(ctx, announce, callParam)
	}
	return err
}

func (p *pool) WaitForContainerPresence(ctx context.Context, cid *cid.ID, pollParams *ContainerPollingParams) error {
	conn, _, err := p.Connection()
	if err != nil {
		return err
	}
	wctx, cancel := context.WithTimeout(ctx, pollParams.CreationTimeout)
	defer cancel()
	ticker := time.NewTimer(pollParams.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:
			_, err = conn.GetContainer(ctx, cid)
			if err == nil {
				return nil
			}
			ticker.Reset(pollParams.PollInterval)
		}
	}
}

// Cloce closes the pool and releases all the associated resources.
func (p *pool) Close() {
	p.cancel()
	<-p.closedCh
}