package replicator import ( "context" "git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/engine" objectwriter "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/object/common/writer" tracingPkg "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/tracing" "git.frostfs.info/TrueCloudLab/frostfs-observability/tracing" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" "go.uber.org/zap" ) // TaskResult is a replication result interface. type TaskResult interface { // SubmitSuccessfulReplication submits the successful object replication // to the given node. SubmitSuccessfulReplication(netmap.NodeInfo) } // HandleReplicationTask executes replication task inside invoking goroutine. // Passes all the nodes that accepted the replication to the TaskResult. func (p *Replicator) HandleReplicationTask(ctx context.Context, task Task, res TaskResult) { p.metrics.IncInFlightRequest() defer p.metrics.DecInFlightRequest() defer func() { p.log.Debug(logs.ReplicatorFinishWork, zap.Uint32("amount of unfinished replicas", task.NumCopies), ) }() ctx, span := tracing.StartSpanFromContext(ctx, "Replicator.HandleReplicateTask", trace.WithAttributes( attribute.Stringer("address", task.Addr), attribute.Int64("number_of_copies", int64(task.NumCopies)), )) defer span.End() if task.Obj == nil { var err error task.Obj, err = engine.Get(ctx, p.localStorage, task.Addr) if err != nil { p.log.Error(logs.ReplicatorCouldNotGetObjectFromLocalStorage, zap.Stringer("object", task.Addr), zap.Error(err), zap.String("trace_id", tracingPkg.GetTraceID(ctx))) return } } prm := new(objectwriter.RemotePutPrm). WithObject(task.Obj) for i := 0; task.NumCopies > 0 && i < len(task.Nodes); i++ { select { case <-ctx.Done(): return default: } log := p.log.With( zap.String("node", netmap.StringifyPublicKey(task.Nodes[i])), zap.Stringer("object", task.Addr), zap.String("trace_id", tracingPkg.GetTraceID(ctx)), ) callCtx, cancel := context.WithTimeout(ctx, p.putTimeout) err := p.remoteSender.PutObject(callCtx, prm.WithNodeInfo(task.Nodes[i])) cancel() if err != nil { log.Error(logs.ReplicatorCouldNotReplicateObject, zap.String("error", err.Error()), ) } else { log.Debug(logs.ReplicatorObjectSuccessfullyReplicated) task.NumCopies-- res.SubmitSuccessfulReplication(task.Nodes[i]) p.metrics.IncProcessedObjects() p.metrics.AddPayloadSize(int64(task.Obj.PayloadSize())) } } }