frostfs-node/pkg/services/tree/replicator.go

package tree

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"time"

	"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/pilorama"
	tracingPkg "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/tracing"
	"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
	cidSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
	netmapSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/zap"
)

type movePair struct {
	cid    cidSDK.ID
	treeID string
	op     *pilorama.Move
}

type replicationTask struct {
	n   netmapSDK.NodeInfo
	req *ApplyRequest
}

type applyOp struct {
	treeID string
	cid    cidSDK.ID
	pilorama.Move
}

const (
	defaultReplicatorCapacity    = 64
	defaultReplicatorWorkerCount = 64
	defaultReplicatorSendTimeout = time.Second * 5
	defaultSyncBatchSize         = 1000
)

func (s *Service) localReplicationWorker(ctx context.Context) {
	for {
		select {
		case <-s.closeCh:
			return
		case op := <-s.replicateLocalCh:
			ctx, span := tracing.StartSpanFromContext(ctx, "TreeService.HandleReplicationOperation",
				trace.WithAttributes(
					attribute.String("tree_id", op.treeID),
					attribute.String("container_id", op.cid.EncodeToString()),
				),
			)

			err := s.forest.TreeApply(ctx, op.cid, op.treeID, &op.Move, false)
			if err != nil {
				s.log.Error(ctx, logs.TreeFailedToApplyReplicatedOperation,
					zap.String("err", err.Error()))
			}
			span.End()
		}
	}
}

func (s *Service) replicationWorker(ctx context.Context) {
	for {
		select {
		case <-s.closeCh:
			return
		case task := <-s.replicationTasks:
			_ = s.ReplicateTreeOp(ctx, task.n, task.req)
		}
	}
}

func (s *Service) ReplicateTreeOp(ctx context.Context, n netmapSDK.NodeInfo, req *ApplyRequest) error {
	ctx, span := tracing.StartSpanFromContext(ctx, "TreeService.HandleReplicationTask",
		trace.WithAttributes(
			attribute.String("public_key", hex.EncodeToString(n.PublicKey())),
		),
	)
	defer span.End()

	start := time.Now()

	var lastErr error
	var lastAddr string

	n.IterateNetworkEndpoints(func(addr string) bool {
		ctx, span := tracing.StartSpanFromContext(ctx, "TreeService.HandleReplicationTaskOnEndpoint",
			trace.WithAttributes(
				attribute.String("public_key", hex.EncodeToString(n.PublicKey())),
				attribute.String("address", addr),
			),
		)
		defer span.End()

		lastAddr = addr

		c, err := s.cache.get(ctx, addr)
		if err != nil {
			lastErr = fmt.Errorf("can't create client: %w", err)
			return false
		}

		ctx, cancel := context.WithTimeout(ctx, s.replicatorTimeout)
		_, lastErr = c.Apply(ctx, req)
		cancel()

		return lastErr == nil
	})

	if lastErr != nil {
		if errors.Is(lastErr, errRecentlyFailed) {
			s.log.Debug(ctx, logs.TreeDoNotSendUpdateToTheNode,
				zap.String("last_error", lastErr.Error()),
				zap.String("trace_id", tracingPkg.GetTraceID(ctx)))
		} else {
			s.log.Warn(ctx, logs.TreeFailedToSentUpdateToTheNode,
				zap.String("last_error", lastErr.Error()),
				zap.String("address", lastAddr),
				zap.String("key", hex.EncodeToString(n.PublicKey())),
				zap.String("trace_id", tracingPkg.GetTraceID(ctx)))
		}
		s.metrics.AddReplicateTaskDuration(time.Since(start), false)
		return lastErr
	}
	s.metrics.AddReplicateTaskDuration(time.Since(start), true)
	return nil
}

func (s *Service) replicateLoop(ctx context.Context) {
	for range s.replicatorWorkerCount {
		go s.replicationWorker(ctx)
		go s.localReplicationWorker(ctx)
	}
	defer func() {
		for len(s.replicationTasks) != 0 {
			<-s.replicationTasks
		}
	}()

	for {
		select {
		case <-s.closeCh:
			return
		case <-ctx.Done():
			return
		case op := <-s.replicateCh:
			start := time.Now()
			err := s.replicate(op)
			if err != nil {
				s.log.Error(ctx, logs.TreeErrorDuringReplication,
					zap.String("err", err.Error()),
					zap.Stringer("cid", op.cid),
					zap.String("treeID", op.treeID))
			}
			s.metrics.AddReplicateWaitDuration(time.Since(start), err == nil)
		}
	}
}

func (s *Service) replicate(op movePair) error {
	req := newApplyRequest(&op)
	err := SignMessage(req, s.key)
	if err != nil {
		return fmt.Errorf("can't sign data: %w", err)
	}

	nodes, localIndex, err := s.getContainerNodes(op.cid)
	if err != nil {
		return fmt.Errorf("can't get container nodes: %w", err)
	}

	for i := range nodes {
		if i != localIndex {
			s.replicationTasks <- replicationTask{nodes[i], req}
		}
	}
	return nil
}

func (s *Service) pushToQueue(cid cidSDK.ID, treeID string, op *pilorama.Move) {
	select {
	case s.replicateCh <- movePair{
		cid:    cid,
		treeID: treeID,
		op:     op,
	}:
	default:
	}
}

func newApplyRequest(op *movePair) *ApplyRequest {
	rawCID := make([]byte, sha256.Size)
	op.cid.Encode(rawCID)

	return &ApplyRequest{
		Body: &ApplyRequest_Body{
			ContainerId: rawCID,
			TreeId:      op.treeID,
			Operation: &LogMove{
				ParentId: op.op.Parent,
				Meta:     op.op.Meta.Bytes(),
				ChildId:  op.op.Child,
			},
		},
	}
}