frostfs-node/pkg/services/object/patch/streamer.go
Airat Arifullin 61d5e140e0
All checks were successful
DCO action / DCO (pull_request) Successful in 58s
Tests and linters / Run gofumpt (pull_request) Successful in 1m15s
Vulncheck / Vulncheck (pull_request) Successful in 1m26s
Pre-commit hooks / Pre-commit (pull_request) Successful in 2m6s
Build / Build Components (pull_request) Successful in 2m20s
Tests and linters / gopls check (pull_request) Successful in 2m34s
Tests and linters / Staticcheck (pull_request) Successful in 2m49s
Tests and linters / Lint (pull_request) Successful in 3m41s
Tests and linters / Tests (pull_request) Successful in 4m46s
Tests and linters / Tests with -race (pull_request) Successful in 4m58s
[#1383] object: Add restrictions for Patch method
* `Patch` can't be applied for non-regular type object (tombstones,
  locks etc.)
* Complex object parts can't be patched. So, if an object has EC/Split
  header, it won't be patched.

Signed-off-by: Airat Arifullin <a.arifullin@yadro.com>
2024-09-19 13:54:49 +03:00

236 lines
5.8 KiB
Go

package patchsvc
import (
"context"
"crypto/ecdsa"
"errors"
"fmt"
"io"
objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
refsV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/refs"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/common/target"
objectwriter "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/common/writer"
getsvc "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/get"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/util"
"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
objectSDK "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/patcher"
)
// Streamer for the patch handler is a pipeline that merges two incoming streams of patches
// and original object payload chunks. The merged result is fed to Put stream target.
type Streamer struct {
*objectwriter.Config
// Patcher must be initialized at first Streamer.Send call.
patcher patcher.PatchApplier
nonFirstSend bool
getSvc *getsvc.Service
localNodeKey *ecdsa.PrivateKey
}
type pipeChunkWriter struct {
wr *io.PipeWriter
}
type headResponseWriter struct {
body *objectV2.HeadResponseBody
}
func (w *headResponseWriter) WriteHeader(_ context.Context, hdr *objectSDK.Object) error {
w.body.SetHeaderPart(toFullObjectHeader(hdr))
return nil
}
func toFullObjectHeader(hdr *objectSDK.Object) objectV2.GetHeaderPart {
obj := hdr.ToV2()
hs := new(objectV2.HeaderWithSignature)
hs.SetHeader(obj.GetHeader())
hs.SetSignature(obj.GetSignature())
return hs
}
func isLinkObject(hdr *objectV2.HeaderWithSignature) bool {
split := hdr.GetHeader().GetSplit()
return len(split.GetChildren()) > 0 && split.GetParent() != nil
}
func isComplexObjectPart(hdr *objectV2.HeaderWithSignature) bool {
return hdr.GetHeader().GetEC() != nil || hdr.GetHeader().GetSplit() != nil
}
func (s *Streamer) init(ctx context.Context, req *objectV2.PatchRequest) error {
hdrWithSig, addr, err := s.readHeader(ctx, req)
if err != nil {
return err
}
if hdrWithSig.GetHeader().GetObjectType() != objectV2.TypeRegular {
return errors.New("non-regular object can't be patched")
}
if isLinkObject(hdrWithSig) {
return errors.New("linking object can't be patched")
}
if isComplexObjectPart(hdrWithSig) {
return errors.New("complex object parts can't be patched")
}
commonPrm, err := util.CommonPrmFromV2(req)
if err != nil {
return err
}
commonPrm.WithLocalOnly(false)
rangeProvider := &rangeProvider{
getSvc: s.getSvc,
addr: addr,
commonPrm: commonPrm,
localNodeKey: s.localNodeKey,
}
hdr := hdrWithSig.GetHeader()
oV2 := new(objectV2.Object)
hV2 := new(objectV2.Header)
oV2.SetHeader(hV2)
oV2.GetHeader().SetContainerID(hdr.GetContainerID())
oV2.GetHeader().SetPayloadLength(hdr.GetPayloadLength())
oV2.GetHeader().SetAttributes(hdr.GetAttributes())
ownerID, err := newOwnerID(req.GetVerificationHeader())
if err != nil {
return err
}
oV2.GetHeader().SetOwnerID(ownerID)
target, err := target.New(&objectwriter.Params{
Config: s.Config,
Common: commonPrm,
Header: objectSDK.NewFromV2(oV2),
SignRequestPrivateKey: s.localNodeKey,
})
if err != nil {
return fmt.Errorf("target creation: %w", err)
}
patcherPrm := patcher.Params{
Header: objectSDK.NewFromV2(oV2),
RangeProvider: rangeProvider,
ObjectWriter: target,
}
s.patcher = patcher.New(patcherPrm)
return nil
}
func (s *Streamer) readHeader(ctx context.Context, req *objectV2.PatchRequest) (hdrWithSig *objectV2.HeaderWithSignature, addr oid.Address, err error) {
addrV2 := req.GetBody().GetAddress()
if addrV2 == nil {
err = errors.New("patch request has nil-address")
return
}
if err = addr.ReadFromV2(*addrV2); err != nil {
err = fmt.Errorf("read address error: %w", err)
return
}
commonPrm, err := util.CommonPrmFromV2(req)
if err != nil {
return
}
commonPrm.WithLocalOnly(false)
var p getsvc.HeadPrm
p.SetSignerKey(s.localNodeKey)
p.SetCommonParameters(commonPrm)
resp := new(objectV2.HeadResponse)
resp.SetBody(new(objectV2.HeadResponseBody))
p.WithAddress(addr)
p.SetHeaderWriter(&headResponseWriter{
body: resp.GetBody(),
})
err = s.getSvc.Head(ctx, p)
if err != nil {
err = fmt.Errorf("get header error: %w", err)
return
}
var ok bool
hdrPart := resp.GetBody().GetHeaderPart()
if hdrWithSig, ok = hdrPart.(*objectV2.HeaderWithSignature); !ok {
err = fmt.Errorf("unexpected header type: %T", hdrPart)
}
return
}
func (s *Streamer) Send(ctx context.Context, req *objectV2.PatchRequest) error {
ctx, span := tracing.StartSpanFromContext(ctx, "patch.streamer.Send")
defer span.End()
defer func() {
s.nonFirstSend = true
}()
if !s.nonFirstSend {
if err := s.init(ctx, req); err != nil {
return fmt.Errorf("streamer init error: %w", err)
}
}
patch := new(objectSDK.Patch)
patch.FromV2(req.GetBody())
if !s.nonFirstSend {
err := s.patcher.ApplyAttributesPatch(ctx, patch.NewAttributes, patch.ReplaceAttributes)
if err != nil {
return fmt.Errorf("patch attributes: %w", err)
}
}
if patch.PayloadPatch != nil {
err := s.patcher.ApplyPayloadPatch(ctx, patch.PayloadPatch)
if err != nil {
return fmt.Errorf("patch payload: %w", err)
}
} else if s.nonFirstSend {
return errors.New("invalid non-first patch: empty payload")
}
return nil
}
func (s *Streamer) CloseAndRecv(ctx context.Context) (*objectV2.PatchResponse, error) {
patcherResp, err := s.patcher.Close(ctx)
if err != nil {
return nil, err
}
oidV2 := new(refsV2.ObjectID)
if patcherResp.AccessIdentifiers.ParentID != nil {
patcherResp.AccessIdentifiers.ParentID.WriteToV2(oidV2)
} else {
patcherResp.AccessIdentifiers.SelfID.WriteToV2(oidV2)
}
return &objectV2.PatchResponse{
Body: &objectV2.PatchResponseBody{
ObjectID: oidV2,
},
}, nil
}