frostfs-node/pkg/services/object/search/container.go

package searchsvc

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

	"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/client"
	containerSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
	"go.uber.org/zap"
)

func (exec *execCtx) executeOnContainer(ctx context.Context) error {
	lookupDepth := exec.netmapLookupDepth()

	exec.log.Debug(ctx, logs.TryingToExecuteInContainer,
		zap.Uint64("netmap lookup depth", lookupDepth),
	)

	// initialize epoch number
	if err := exec.initEpoch(); err != nil {
		return fmt.Errorf("%s: %w", logs.CouldNotGetCurrentEpochNumber, err)
	}

	for {
		if err := exec.processCurrentEpoch(ctx); err != nil {
			break
		}

		// check the maximum depth has been reached
		if lookupDepth == 0 {
			break
		}

		lookupDepth--

		// go to the previous epoch
		exec.curProcEpoch--
	}

	return nil
}

func (exec *execCtx) processCurrentEpoch(ctx context.Context) error {
	exec.log.Debug(ctx, logs.ProcessEpoch,
		zap.Uint64("number", exec.curProcEpoch),
	)

	traverser, _, err := exec.svc.traverserGenerator.GenerateTraverser(exec.containerID(), nil, exec.curProcEpoch)
	if err != nil {
		return fmt.Errorf("%s: %w", logs.SearchCouldNotGenerateContainerTraverser, err)
	}

	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	for {
		addrs := traverser.Next()
		if len(addrs) == 0 {
			exec.log.Debug(ctx, logs.NoMoreNodesAbortPlacementIteration)
			break
		}

		var wg sync.WaitGroup
		var mtx sync.Mutex

		for i := range addrs {
			wg.Add(1)
			go func(i int) {
				defer wg.Done()
				select {
				case <-ctx.Done():
					exec.log.Debug(ctx, logs.InterruptPlacementIterationByContext,
						zap.String("error", ctx.Err().Error()))
					return
				default:
				}

				var info client.NodeInfo

				client.NodeInfoFromNetmapElement(&info, addrs[i])

				exec.log.Debug(ctx, logs.ProcessingNode, zap.String("key", hex.EncodeToString(addrs[i].PublicKey())))

				c, err := exec.svc.clientConstructor.get(info)
				if err != nil {
					exec.log.Debug(ctx, logs.SearchCouldNotConstructRemoteNodeClient, zap.String("error", err.Error()))
					return
				}

				ids, err := c.searchObjects(ctx, exec, info)
				if err != nil {
					exec.log.Debug(ctx, logs.SearchRemoteOperationFailed,
						zap.String("error", err.Error()))

					return
				}

				mtx.Lock()
				err = exec.writeIDList(ids)
				mtx.Unlock()
				if err != nil {
					exec.log.Debug(ctx, logs.SearchCouldNotWriteObjectIdentifiers, zap.String("error", err.Error()))
					return
				}
			}(i)
		}

		wg.Wait()
	}

	return nil
}

func (exec *execCtx) getContainer() (containerSDK.Container, error) {
	cnrID := exec.containerID()
	cnr, err := exec.svc.containerSource.Get(cnrID)
	if err != nil {
		return containerSDK.Container{}, err
	}
	return cnr.Value, nil
}