157 lines
3.8 KiB
Go
157 lines
3.8 KiB
Go
package putsvc
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object_manager/placement"
|
|
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/transformer"
|
|
)
|
|
|
|
type preparedObjectTarget interface {
|
|
WriteObject(context.Context, *objectSDK.Object, object.ContentMeta) error
|
|
}
|
|
|
|
type distributedTarget struct {
|
|
placementOpts []placement.Option
|
|
extraBroadcastEnabled bool
|
|
|
|
obj *objectSDK.Object
|
|
objMeta object.ContentMeta
|
|
|
|
*cfg
|
|
|
|
payload *payload
|
|
|
|
nodeTargetInitializer func(nodeDesc) preparedObjectTarget
|
|
|
|
relay func(context.Context, nodeDesc) error
|
|
}
|
|
|
|
// parameters and state of container traversal.
|
|
type traversal struct {
|
|
opts []placement.Option
|
|
|
|
// need of additional broadcast after the object is saved
|
|
extraBroadcastEnabled bool
|
|
|
|
// container nodes which was processed during the primary object placement
|
|
mExclude map[string]*bool
|
|
}
|
|
|
|
// updates traversal parameters after the primary placement finish and
|
|
// returns true if additional container broadcast is needed.
|
|
func (x *traversal) submitPrimaryPlacementFinish() bool {
|
|
if x.extraBroadcastEnabled {
|
|
// do not track success during container broadcast (best-effort)
|
|
x.opts = append(x.opts, placement.WithoutSuccessTracking())
|
|
|
|
// avoid 2nd broadcast
|
|
x.extraBroadcastEnabled = false
|
|
|
|
return true
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
// marks the container node as processed during the primary object placement.
|
|
func (x *traversal) submitProcessed(n placement.Node, item *bool) {
|
|
if x.extraBroadcastEnabled {
|
|
key := string(n.PublicKey())
|
|
|
|
if x.mExclude == nil {
|
|
x.mExclude = make(map[string]*bool, 1)
|
|
}
|
|
|
|
x.mExclude[key] = item
|
|
}
|
|
}
|
|
|
|
type nodeDesc struct {
|
|
local bool
|
|
|
|
info placement.Node
|
|
}
|
|
|
|
// errIncompletePut is returned if processing on a container fails.
|
|
type errIncompletePut struct {
|
|
singleErr error // error from the last responding node
|
|
}
|
|
|
|
func (x errIncompletePut) Error() string {
|
|
const commonMsg = "incomplete object PUT by placement"
|
|
|
|
if x.singleErr != nil {
|
|
return fmt.Sprintf("%s: %v", commonMsg, x.singleErr)
|
|
}
|
|
|
|
return commonMsg
|
|
}
|
|
|
|
func (t *distributedTarget) WriteHeader(_ context.Context, obj *objectSDK.Object) error {
|
|
t.obj = obj
|
|
|
|
return nil
|
|
}
|
|
|
|
func (t *distributedTarget) Write(_ context.Context, p []byte) (n int, err error) {
|
|
t.payload.Data = append(t.payload.Data, p...)
|
|
|
|
return len(p), nil
|
|
}
|
|
|
|
func (t *distributedTarget) Close(ctx context.Context) (*transformer.AccessIdentifiers, error) {
|
|
defer func() {
|
|
putPayload(t.payload)
|
|
t.payload = nil
|
|
}()
|
|
|
|
t.obj.SetPayload(t.payload.Data)
|
|
|
|
if err := t.WriteObject(ctx, t.obj); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
id, _ := t.obj.ID()
|
|
return &transformer.AccessIdentifiers{
|
|
SelfID: id,
|
|
}, nil
|
|
}
|
|
|
|
// WriteObject implements the transformer.ObjectWriter interface.
|
|
func (t *distributedTarget) WriteObject(ctx context.Context, obj *objectSDK.Object) error {
|
|
t.obj = obj
|
|
|
|
var err error
|
|
|
|
if t.objMeta, err = t.fmtValidator.ValidateContent(t.obj); err != nil {
|
|
return fmt.Errorf("(%T) could not validate payload content: %w", t, err)
|
|
}
|
|
return t.iteratePlacement(ctx)
|
|
}
|
|
|
|
func (t *distributedTarget) sendObject(ctx context.Context, node nodeDesc) error {
|
|
if !node.local && t.relay != nil {
|
|
return t.relay(ctx, node)
|
|
}
|
|
|
|
target := t.nodeTargetInitializer(node)
|
|
|
|
err := target.WriteObject(ctx, t.obj, t.objMeta)
|
|
if err != nil {
|
|
return fmt.Errorf("could not write header: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (t *distributedTarget) iteratePlacement(ctx context.Context) error {
|
|
id, _ := t.obj.ID()
|
|
|
|
iter := t.cfg.newNodeIterator(append(t.placementOpts, placement.ForObject(id)))
|
|
iter.extraBroadcastEnabled = needAdditionalBroadcast(t.obj, false /* Distributed target is for cluster-wide PUT */)
|
|
return iter.forEachNode(ctx, t.sendObject)
|
|
}
|