frostfs-node/pkg/services/object/put/remote.go

package putsvc

import (
	"context"
	"crypto/ecdsa"
	"fmt"

	clientcore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/client"
	netmapCore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/netmap"
	objectcore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
	internalclient "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/internal/client"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/util"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
	objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/transformer"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

type remoteTarget struct {
	privateKey *ecdsa.PrivateKey

	commonPrm *util.CommonPrm

	nodeInfo clientcore.NodeInfo

	obj *objectSDK.Object

	clientConstructor ClientConstructor
}

// RemoteSender represents utility for
// sending an object to a remote host.
type RemoteSender struct {
	keyStorage *util.KeyStorage

	clientConstructor ClientConstructor
}

// RemotePutPrm groups remote put operation parameters.
type RemotePutPrm struct {
	node netmap.NodeInfo

	obj *objectSDK.Object
}

func (t *remoteTarget) WriteObject(obj *objectSDK.Object, _ objectcore.ContentMeta) error {
	t.obj = obj

	return nil
}

func (t *remoteTarget) Close(ctx context.Context) (*transformer.AccessIdentifiers, error) {
	c, err := t.clientConstructor.Get(t.nodeInfo)
	if err != nil {
		return nil, fmt.Errorf("(%T) could not create SDK client %s: %w", t, t.nodeInfo, err)
	}

	var prm internalclient.PutObjectPrm

	prm.SetClient(c)
	prm.SetPrivateKey(t.privateKey)
	prm.SetSessionToken(t.commonPrm.SessionToken())
	prm.SetBearerToken(t.commonPrm.BearerToken())
	prm.SetXHeaders(t.commonPrm.XHeaders())
	prm.SetObject(t.obj)

	res, err := t.putSingle(ctx, prm)
	if status.Code(err) != codes.Unimplemented {
		return res, err
	}

	return t.putStream(ctx, prm)
}

func (t *remoteTarget) putStream(ctx context.Context, prm internalclient.PutObjectPrm) (*transformer.AccessIdentifiers, error) {
	res, err := internalclient.PutObject(ctx, prm)
	if err != nil {
		return nil, fmt.Errorf("(%T) could not put object to %s: %w", t, t.nodeInfo.AddressGroup(), err)
	}

	return &transformer.AccessIdentifiers{SelfID: res.ID()}, nil
}

func (t *remoteTarget) putSingle(ctx context.Context, prm internalclient.PutObjectPrm) (*transformer.AccessIdentifiers, error) {
	res, err := internalclient.PutObjectSingle(ctx, prm)
	if err != nil {
		return nil, fmt.Errorf("(%T) could not put single object to %s: %w", t, t.nodeInfo.AddressGroup(), err)
	}

	return &transformer.AccessIdentifiers{SelfID: res.ID()}, nil
}

// NewRemoteSender creates, initializes and returns new RemoteSender instance.
func NewRemoteSender(keyStorage *util.KeyStorage, cons ClientConstructor) *RemoteSender {
	return &RemoteSender{
		keyStorage:        keyStorage,
		clientConstructor: cons,
	}
}

// WithNodeInfo sets information about the remote node.
func (p *RemotePutPrm) WithNodeInfo(v netmap.NodeInfo) *RemotePutPrm {
	if p != nil {
		p.node = v
	}

	return p
}

// WithObject sets transferred object.
func (p *RemotePutPrm) WithObject(v *objectSDK.Object) *RemotePutPrm {
	if p != nil {
		p.obj = v
	}

	return p
}

// PutObject sends object to remote node.
func (s *RemoteSender) PutObject(ctx context.Context, p *RemotePutPrm) error {
	key, err := s.keyStorage.GetKey(nil)
	if err != nil {
		return err
	}

	t := &remoteTarget{
		privateKey:        key,
		clientConstructor: s.clientConstructor,
	}

	err = clientcore.NodeInfoFromRawNetmapElement(&t.nodeInfo, netmapCore.Node(p.node))
	if err != nil {
		return fmt.Errorf("parse client node info: %w", err)
	}

	if err := t.WriteObject(p.obj, objectcore.ContentMeta{}); err != nil {
		return fmt.Errorf("(%T) could not send object header: %w", s, err)
	} else if _, err := t.Close(ctx); err != nil {
		return fmt.Errorf("(%T) could not send object: %w", s, err)
	}

	return nil
}