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) }