Evgenii Stratonikov
22be532cbd
Previously a token could've expired in the middle of an object.PUT stream, leading to upload being interrupted. This is bad, because user doesn't always now what is the right values for the session token lifetime. More than that, setting it to a very high value will eventually blow up the session token database. In this commit we read the session token once and reuse it for the whole stream duration. Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
129 lines
3.2 KiB
Go
129 lines
3.2 KiB
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-node/pkg/services/object_manager/transformer"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
|
)
|
|
|
|
type remoteTarget struct {
|
|
ctx context.Context
|
|
|
|
privateKey *ecdsa.PrivateKey
|
|
|
|
commonPrm *util.CommonPrm
|
|
|
|
nodeInfo clientcore.NodeInfo
|
|
|
|
obj *object.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 *object.Object
|
|
}
|
|
|
|
func (t *remoteTarget) WriteObject(obj *object.Object, _ objectcore.ContentMeta) error {
|
|
t.obj = obj
|
|
|
|
return nil
|
|
}
|
|
|
|
func (t *remoteTarget) Close() (*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.SetContext(t.ctx)
|
|
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 := internalclient.PutObject(prm)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("(%T) could not put object to %s: %w", t, t.nodeInfo.AddressGroup(), err)
|
|
}
|
|
|
|
return new(transformer.AccessIdentifiers).
|
|
WithSelfID(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 *object.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{
|
|
ctx: ctx,
|
|
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(); err != nil {
|
|
return fmt.Errorf("(%T) could not send object: %w", s, err)
|
|
}
|
|
|
|
return nil
|
|
}
|