frostfs-node/pkg/services/object/common/writer/distributed.go

132 lines
3.3 KiB
Go
Raw Normal View History

package writer
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"
)
type preparedObjectTarget interface {
WriteObject(context.Context, *objectSDK.Object, object.ContentMeta) error
}
type distributedWriter struct {
cfg *Config
placementOpts []placement.Option
obj *objectSDK.Object
objMeta object.ContentMeta
nodeTargetInitializer func(NodeDescriptor) preparedObjectTarget
relay func(context.Context, NodeDescriptor) error
resetSuccessAfterOnBroadcast bool
}
// 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
Exclude map[string]*bool
ResetSuccessAfterOnBroadcast 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())
if x.ResetSuccessAfterOnBroadcast {
x.Opts = append(x.Opts, placement.ResetSuccessAfter())
}
// 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.Exclude == nil {
x.Exclude = make(map[string]*bool, 1)
}
x.Exclude[key] = item
}
}
type NodeDescriptor 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
}
// WriteObject implements the transformer.ObjectWriter interface.
func (t *distributedWriter) WriteObject(ctx context.Context, obj *objectSDK.Object) error {
t.obj = obj
var err error
if t.objMeta, err = t.cfg.FormatValidator.ValidateContent(t.obj); err != nil {
return fmt.Errorf("(%T) could not validate payload content: %w", t, err)
}
return t.iteratePlacement(ctx)
}
func (t *distributedWriter) sendObject(ctx context.Context, node NodeDescriptor) 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 *distributedWriter) 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 */)
iter.ResetSuccessAfterOnBroadcast = t.resetSuccessAfterOnBroadcast
return iter.ForEachNode(ctx, t.sendObject)
}