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

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
}

// Traversal parameters and state of container.
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)
}