frostfs-node/pkg/services/policer/ec.go

package policer

import (
	"context"
	"encoding/hex"
	"errors"
	"fmt"

	"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
	objectcore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/replicator"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
	objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/erasurecode"
	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
)

var errNoECinfoReturnded = errors.New("no EC info returned")

type ecChunkProcessResult struct {
	validPlacement bool
	removeLocal    bool
}

var errInvalidECPlacement = errors.New("invalid EC placement: EC placement must have one placement vector")

func (p *Policer) processECContainerObject(ctx context.Context, objInfo objectcore.Info, policy netmap.PlacementPolicy) error {
	if objInfo.ECInfo == nil {
		return p.processECContainerRepObject(ctx, objInfo, policy)
	}
	return p.processECContainerECObject(ctx, objInfo, policy)
}

// processECContainerRepObject processes non erasure coded objects in EC container: tombstones, locks and linking objects.
// All of them must be stored on all of the container nodes.
func (p *Policer) processECContainerRepObject(ctx context.Context, objInfo objectcore.Info, policy netmap.PlacementPolicy) error {
	objID := objInfo.Address.Object()
	nn, err := p.placementBuilder.BuildPlacement(objInfo.Address.Container(), &objID, policy)
	if err != nil {
		return fmt.Errorf("%s: %w", logs.PolicerCouldNotBuildPlacementVectorForObject, err)
	}
	if len(nn) != 1 {
		return errInvalidECPlacement
	}

	c := &placementRequirements{}
	checkedNodes := newNodeCache()

	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
	}

	p.processRepNodes(ctx, c, objInfo, nn[0], uint32(len(nn[0])), checkedNodes)

	if !c.needLocalCopy && c.removeLocalCopy {
		p.log.Info(logs.PolicerRedundantLocalObjectCopyDetected,
			zap.Stringer("object", objInfo.Address),
		)

		p.cbRedundantCopy(ctx, objInfo.Address)
	}
	return nil
}

func (p *Policer) processECContainerECObject(ctx context.Context, objInfo objectcore.Info, policy netmap.PlacementPolicy) error {
	nn, err := p.placementBuilder.BuildPlacement(objInfo.Address.Container(), &objInfo.ECInfo.ParentID, policy)
	if err != nil {
		return fmt.Errorf("%s: %w", logs.PolicerCouldNotBuildPlacementVectorForObject, err)
	}
	if len(nn) != 1 {
		return errInvalidECPlacement
	}

	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
	}

	res := p.processECChunk(ctx, objInfo, nn[0])
	if !res.validPlacement {
		// drop local chunk only if all required chunks are in place
		res.removeLocal = res.removeLocal && p.pullRequiredECChunks(ctx, objInfo, nn[0])
	}
	p.adjustECPlacement(ctx, objInfo, nn[0], policy)

	if res.removeLocal {
		p.log.Info(logs.PolicerRedundantLocalObjectCopyDetected, zap.Stringer("object", objInfo.Address))
		p.cbRedundantCopy(ctx, objInfo.Address)
	}
	return nil
}

// processECChunk replicates EC chunk if needed.
func (p *Policer) processECChunk(ctx context.Context, objInfo objectcore.Info, nodes []netmap.NodeInfo) ecChunkProcessResult {
	var removeLocalChunk bool
	requiredNode := nodes[int(objInfo.ECInfo.Index)%(len(nodes))]
	if p.cfg.netmapKeys.IsLocalKey(requiredNode.PublicKey()) {
		// current node is required node, we are happy
		return ecChunkProcessResult{
			validPlacement: true,
		}
	}
	if requiredNode.IsMaintenance() {
		// consider maintenance mode has object, but do not drop local copy
		p.log.Debug(logs.PolicerConsiderNodeUnderMaintenanceAsOK, zap.String("node", netmap.StringifyPublicKey(requiredNode)))
		return ecChunkProcessResult{}
	}

	callCtx, cancel := context.WithTimeout(ctx, p.headTimeout)
	_, err := p.remoteHeader(callCtx, requiredNode, objInfo.Address, false)
	cancel()

	if err == nil {
		removeLocalChunk = true
	} else if client.IsErrObjectNotFound(err) {
		p.log.Debug(logs.PolicerShortageOfObjectCopiesDetected, zap.Stringer("object", objInfo.Address), zap.Uint32("shortage", 1))
		task := replicator.Task{
			NumCopies: 1,
			Addr:      objInfo.Address,
			Nodes:     []netmap.NodeInfo{requiredNode},
		}
		p.replicator.HandleReplicationTask(ctx, task, newNodeCache())
	} else if isClientErrMaintenance(err) {
		// consider maintenance mode has object, but do not drop local copy
		p.log.Debug(logs.PolicerConsiderNodeUnderMaintenanceAsOK, zap.String("node", netmap.StringifyPublicKey(requiredNode)))
	} else {
		p.log.Error(logs.PolicerReceiveObjectHeaderToCheckPolicyCompliance, zap.Stringer("object", objInfo.Address), zap.String("error", err.Error()))
	}

	return ecChunkProcessResult{
		removeLocal: removeLocalChunk,
	}
}

func (p *Policer) pullRequiredECChunks(ctx context.Context, objInfo objectcore.Info, nodes []netmap.NodeInfo) bool {
	var parentAddress oid.Address
	parentAddress.SetContainer(objInfo.Address.Container())
	parentAddress.SetObject(objInfo.ECInfo.ParentID)

	requiredChunkIndexes := p.collectRequiredECChunks(nodes, objInfo)
	if len(requiredChunkIndexes) == 0 {
		p.log.Info(logs.PolicerNodeIsNotECObjectNode, zap.Stringer("object", objInfo.ECInfo.ParentID))
		return true
	}

	err := p.resolveLocalECChunks(ctx, parentAddress, requiredChunkIndexes)
	if err != nil {
		p.log.Error(logs.PolicerFailedToGetLocalECChunks, zap.Error(err), zap.Stringer("object", parentAddress))
		return false
	}
	if len(requiredChunkIndexes) == 0 {
		return true
	}

	indexToObjectID := make(map[uint32]oid.ID)
	success := p.resolveRemoteECChunks(ctx, parentAddress, nodes, requiredChunkIndexes, indexToObjectID)
	if !success {
		return false
	}

	for index, candidates := range requiredChunkIndexes {
		var addr oid.Address
		addr.SetContainer(objInfo.Address.Container())
		addr.SetObject(indexToObjectID[index])
		p.replicator.HandlePullTask(ctx, replicator.Task{
			Addr:  addr,
			Nodes: candidates,
		})
	}
	// there was some missing chunks, it's not ok
	return false
}

func (p *Policer) collectRequiredECChunks(nodes []netmap.NodeInfo, objInfo objectcore.Info) map[uint32][]netmap.NodeInfo {
	requiredChunkIndexes := make(map[uint32][]netmap.NodeInfo)
	for i, n := range nodes {
		if uint32(i) == objInfo.ECInfo.Total {
			break
		}
		if p.cfg.netmapKeys.IsLocalKey(n.PublicKey()) {
			requiredChunkIndexes[uint32(i)] = []netmap.NodeInfo{}
		}
	}
	return requiredChunkIndexes
}

func (p *Policer) resolveLocalECChunks(ctx context.Context, parentAddress oid.Address, required map[uint32][]netmap.NodeInfo) error {
	_, err := p.localHeader(ctx, parentAddress)
	var eiErr *objectSDK.ECInfoError
	if err == nil { // should not be happen
		return errNoECinfoReturnded
	}
	if !errors.As(err, &eiErr) {
		return err
	}
	for _, ch := range eiErr.ECInfo().Chunks {
		delete(required, ch.Index)
	}
	return nil
}

func (p *Policer) resolveRemoteECChunks(ctx context.Context, parentAddress oid.Address, nodes []netmap.NodeInfo, required map[uint32][]netmap.NodeInfo, indexToObjectID map[uint32]oid.ID) bool {
	var eiErr *objectSDK.ECInfoError
	for _, n := range nodes {
		if p.cfg.netmapKeys.IsLocalKey(n.PublicKey()) {
			continue
		}
		_, err := p.remoteHeader(ctx, n, parentAddress, true)
		if !errors.As(err, &eiErr) {
			continue
		}
		for _, ch := range eiErr.ECInfo().Chunks {
			if candidates, ok := required[ch.Index]; ok {
				candidates = append(candidates, n)
				required[ch.Index] = candidates

				var chunkID oid.ID
				if err := chunkID.ReadFromV2(ch.ID); err != nil {
					p.log.Error(logs.PolicerFailedToDecodeECChunkID, zap.Error(err), zap.Stringer("object", parentAddress))
					return false
				}
				if existed, ok := indexToObjectID[ch.Index]; ok && existed != chunkID {
					p.log.Error(logs.PolicerDifferentObjectIDForTheSameECChunk, zap.Stringer("first", existed),
						zap.Stringer("second", chunkID), zap.Stringer("object", parentAddress), zap.Uint32("index", ch.Index))
					return false
				}
				indexToObjectID[ch.Index] = chunkID
			}
		}
	}

	for index, candidates := range required {
		if len(candidates) == 0 {
			p.log.Error(logs.PolicerMissingECChunk, zap.Stringer("object", parentAddress), zap.Uint32("index", index))
			return false
		}
	}

	return true
}

func (p *Policer) adjustECPlacement(ctx context.Context, objInfo objectcore.Info, nodes []netmap.NodeInfo, policy netmap.PlacementPolicy) {
	var parentAddress oid.Address
	parentAddress.SetContainer(objInfo.Address.Container())
	parentAddress.SetObject(objInfo.ECInfo.ParentID)
	var eiErr *objectSDK.ECInfoError
	resolved := make(map[uint32][]netmap.NodeInfo)
	chunkIDs := make(map[uint32]oid.ID)
	restore := true // do not restore EC chunks if some node returned error
	for idx, n := range nodes {
		if uint32(idx) >= objInfo.ECInfo.Total && uint32(len(resolved)) == objInfo.ECInfo.Total {
			return
		}
		var err error
		if p.cfg.netmapKeys.IsLocalKey(n.PublicKey()) {
			_, err = p.localHeader(ctx, parentAddress)
		} else {
			_, err = p.remoteHeader(ctx, n, parentAddress, true)
		}

		if errors.As(err, &eiErr) {
			for _, ch := range eiErr.ECInfo().Chunks {
				resolved[ch.Index] = append(resolved[ch.Index], n)
				var ecInfoChunkID oid.ID
				if err := ecInfoChunkID.ReadFromV2(ch.ID); err != nil {
					p.log.Error(logs.PolicerFailedToDecodeECChunkID, zap.Error(err), zap.Stringer("object", parentAddress))
					return
				}
				if chunkID, exist := chunkIDs[ch.Index]; exist && chunkID != ecInfoChunkID {
					p.log.Error(logs.PolicerDifferentObjectIDForTheSameECChunk, zap.Stringer("first", chunkID),
						zap.Stringer("second", ecInfoChunkID), zap.Stringer("object", parentAddress), zap.Uint32("index", ch.Index))
					return
				}
				chunkIDs[ch.Index] = ecInfoChunkID
			}
		} else if !p.cfg.netmapKeys.IsLocalKey(n.PublicKey()) && uint32(idx) < objInfo.ECInfo.Total {
			p.log.Warn(logs.PolicerCouldNotGetObjectFromNodeMoving, zap.String("node", hex.EncodeToString(n.PublicKey())), zap.Stringer("object", parentAddress), zap.Error(err))
			p.replicator.HandleReplicationTask(ctx, replicator.Task{
				NumCopies: 1,
				Addr:      objInfo.Address,
				Nodes:     []netmap.NodeInfo{n},
			}, newNodeCache())
			restore = false
		}
	}
	if !restore || uint32(len(resolved)) == objInfo.ECInfo.Total {
		return
	}
	if objInfo.ECInfo.Total-uint32(len(resolved)) > policy.ReplicaDescriptor(0).GetECParityCount() {
		var found []uint32
		for i := range resolved {
			found = append(found, i)
		}
		p.log.Error(logs.PolicerCouldNotRestoreObjectNotEnoughChunks, zap.Stringer("object", parentAddress), zap.Uint32s("found_chunks", found))
		return
	}
	p.restoreECObject(ctx, objInfo, parentAddress, nodes, resolved, chunkIDs, policy)
}

func (p *Policer) restoreECObject(ctx context.Context, objInfo objectcore.Info, parentAddress oid.Address, nodes []netmap.NodeInfo, existedChunks map[uint32][]netmap.NodeInfo, chunkIDs map[uint32]oid.ID, policy netmap.PlacementPolicy) {
	c, err := erasurecode.NewConstructor(int(policy.ReplicaDescriptor(0).GetECDataCount()), int(policy.ReplicaDescriptor(0).GetECParityCount()))
	if err != nil {
		p.log.Error(logs.PolicerFailedToRestoreObject, zap.Stringer("object", parentAddress), zap.Error(err))
		return
	}
	parts := p.collectExistedChunks(ctx, objInfo, existedChunks, parentAddress, chunkIDs)
	if parts == nil {
		return
	}
	key, err := p.keyStorage.GetKey(nil)
	if err != nil {
		p.log.Error(logs.PolicerFailedToRestoreObject, zap.Stringer("object", parentAddress), zap.Error(err))
		return
	}
	required := make([]bool, len(parts))
	for i, p := range parts {
		if p == nil {
			required[i] = true
		}
	}
	if err := c.ReconstructParts(parts, required, key); err != nil {
		p.log.Error(logs.PolicerFailedToRestoreObject, zap.Stringer("object", parentAddress), zap.Error(err))
		return
	}
	for idx, part := range parts {
		if _, exists := existedChunks[uint32(idx)]; exists {
			continue
		}
		var addr oid.Address
		addr.SetContainer(parentAddress.Container())
		pID, _ := part.ID()
		addr.SetObject(pID)
		targetNode := nodes[idx%len(nodes)]
		if p.cfg.netmapKeys.IsLocalKey(targetNode.PublicKey()) {
			p.replicator.HandleLocalPutTask(ctx, replicator.Task{
				Addr: addr,
				Obj:  part,
			})
		} else {
			p.replicator.HandleReplicationTask(ctx, replicator.Task{
				NumCopies: 1,
				Addr:      addr,
				Nodes:     []netmap.NodeInfo{targetNode},
				Obj:       part,
			}, newNodeCache())
		}
	}
}

func (p *Policer) collectExistedChunks(ctx context.Context, objInfo objectcore.Info, existedChunks map[uint32][]netmap.NodeInfo, parentAddress oid.Address, chunkIDs map[uint32]oid.ID) []*objectSDK.Object {
	parts := make([]*objectSDK.Object, objInfo.ECInfo.Total)
	errGroup, egCtx := errgroup.WithContext(ctx)
	for idx, nodes := range existedChunks {
		idx := idx
		nodes := nodes
		errGroup.Go(func() error {
			var objID oid.Address
			objID.SetContainer(parentAddress.Container())
			objID.SetObject(chunkIDs[idx])
			var obj *objectSDK.Object
			var err error
			for _, node := range nodes {
				if p.cfg.netmapKeys.IsLocalKey(node.PublicKey()) {
					obj, err = p.localObject(egCtx, objID)
				} else {
					obj, err = p.remoteObject(egCtx, node, objID)
				}
				if err == nil {
					break
				}
				p.log.Warn(logs.PolicerCouldNotGetChunk, zap.Stringer("object", parentAddress), zap.Stringer("chunkID", objID), zap.Error(err), zap.String("node", hex.EncodeToString(node.PublicKey())))
			}
			if obj != nil {
				parts[idx] = obj
			}
			return nil
		})
	}
	if err := errGroup.Wait(); err != nil {
		p.log.Error(logs.PolicerCouldNotGetChunks, zap.Stringer("object", parentAddress), zap.Error(err))
		return nil
	}
	return parts
}