Dmitrii Stepanov
c58ab0c369
All checks were successful
ci/woodpecker/push/pre-commit Pipeline was successful
Get private key only once for request forwaring. Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
168 lines
4.6 KiB
Go
168 lines
4.6 KiB
Go
package getsvc
|
|
|
|
import (
|
|
"context"
|
|
"crypto/ecdsa"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"sync"
|
|
|
|
objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/rpc"
|
|
rpcclient "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/rpc/client"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/session"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/signature"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/client"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/network"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/internal"
|
|
internalclient "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/internal/client"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
|
)
|
|
|
|
type getRequestForwarder struct {
|
|
OnceResign *sync.Once
|
|
OnceHeaderSending *sync.Once
|
|
GlobalProgress int
|
|
Key *ecdsa.PrivateKey
|
|
Request *objectV2.GetRequest
|
|
Stream *streamObjectWriter
|
|
}
|
|
|
|
func (f *getRequestForwarder) forwardRequestToNode(ctx context.Context, addr network.Address, c client.MultiAddressClient, pubkey []byte) (*object.Object, error) {
|
|
var err error
|
|
|
|
// once compose and resign forwarding request
|
|
f.OnceResign.Do(func() {
|
|
// compose meta header of the local server
|
|
metaHdr := new(session.RequestMetaHeader)
|
|
metaHdr.SetTTL(f.Request.GetMetaHeader().GetTTL() - 1)
|
|
// TODO: #1165 think how to set the other fields
|
|
metaHdr.SetOrigin(f.Request.GetMetaHeader())
|
|
writeCurrentVersion(metaHdr)
|
|
f.Request.SetMetaHeader(metaHdr)
|
|
err = signature.SignServiceMessage(f.Key, f.Request)
|
|
})
|
|
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
getStream, err := f.openStream(ctx, addr, c)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return nil, f.readStream(ctx, c, getStream, pubkey)
|
|
}
|
|
|
|
func (f *getRequestForwarder) verifyResponse(resp *objectV2.GetResponse, pubkey []byte) error {
|
|
// verify response key
|
|
if err := internal.VerifyResponseKeyV2(pubkey, resp); err != nil {
|
|
return err
|
|
}
|
|
|
|
// verify response structure
|
|
if err := signature.VerifyServiceMessage(resp); err != nil {
|
|
return fmt.Errorf("response verification failed: %w", err)
|
|
}
|
|
|
|
if err := checkStatus(resp.GetMetaHeader().GetStatus()); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (f *getRequestForwarder) writeHeader(ctx context.Context, v *objectV2.GetObjectPartInit) error {
|
|
obj := new(objectV2.Object)
|
|
|
|
obj.SetObjectID(v.GetObjectID())
|
|
obj.SetSignature(v.GetSignature())
|
|
obj.SetHeader(v.GetHeader())
|
|
|
|
var err error
|
|
f.OnceHeaderSending.Do(func() {
|
|
err = f.Stream.WriteHeader(ctx, object.NewFromV2(obj))
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("could not write object header in Get forwarder: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (f *getRequestForwarder) openStream(ctx context.Context, addr network.Address, c client.MultiAddressClient) (*rpc.GetResponseReader, error) {
|
|
var getStream *rpc.GetResponseReader
|
|
err := c.RawForAddress(addr, func(cli *rpcclient.Client) error {
|
|
var e error
|
|
getStream, e = rpc.GetObject(cli, f.Request, rpcclient.WithContext(ctx))
|
|
return e
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("stream opening failed: %w", err)
|
|
}
|
|
return getStream, nil
|
|
}
|
|
|
|
func (f *getRequestForwarder) readStream(ctx context.Context, c client.MultiAddressClient, getStream *rpc.GetResponseReader, pubkey []byte) error {
|
|
var (
|
|
headWas bool
|
|
resp = new(objectV2.GetResponse)
|
|
localProgress int
|
|
)
|
|
|
|
for {
|
|
// receive message from server stream
|
|
err := getStream.Read(resp)
|
|
if err != nil {
|
|
if errors.Is(err, io.EOF) {
|
|
if !headWas {
|
|
return io.ErrUnexpectedEOF
|
|
}
|
|
|
|
break
|
|
}
|
|
|
|
internalclient.ReportError(c, err)
|
|
return fmt.Errorf("reading the response failed: %w", err)
|
|
}
|
|
|
|
if err := f.verifyResponse(resp, pubkey); err != nil {
|
|
return err
|
|
}
|
|
|
|
switch v := resp.GetBody().GetObjectPart().(type) {
|
|
default:
|
|
return fmt.Errorf("unexpected object part %T", v)
|
|
case *objectV2.GetObjectPartInit:
|
|
if headWas {
|
|
return errWrongMessageSeq
|
|
}
|
|
headWas = true
|
|
if err := f.writeHeader(ctx, v); err != nil {
|
|
return err
|
|
}
|
|
case *objectV2.GetObjectPartChunk:
|
|
if !headWas {
|
|
return errWrongMessageSeq
|
|
}
|
|
|
|
origChunk := v.GetChunk()
|
|
|
|
chunk := chunkToSend(f.GlobalProgress, localProgress, origChunk)
|
|
if len(chunk) == 0 {
|
|
localProgress += len(origChunk)
|
|
continue
|
|
}
|
|
|
|
if err = f.Stream.WriteChunk(ctx, chunk); err != nil {
|
|
return fmt.Errorf("could not write object chunk in Get forwarder: %w", err)
|
|
}
|
|
|
|
localProgress += len(origChunk)
|
|
f.GlobalProgress += len(chunk)
|
|
case *objectV2.SplitInfo:
|
|
si := object.NewSplitInfoFromV2(v)
|
|
return object.NewSplitInfoError(si)
|
|
}
|
|
}
|
|
return nil
|
|
}
|