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

package tree

import (
	"bytes"
	"context"
	"errors"

	"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
	"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
	netmapSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/zap"
	"google.golang.org/grpc"
)

var errNoSuitableNode = errors.New("no node was found to execute the request")

func relayUnary[Req any, Resp any](ctx context.Context, s *Service, ns []netmapSDK.NodeInfo, req *Req, callback func(TreeServiceClient, context.Context, *Req, ...grpc.CallOption) (*Resp, error)) (*Resp, error) {
	var resp *Resp
	var outErr error
	err := s.forEachNode(ctx, ns, func(fCtx context.Context, c TreeServiceClient) bool {
		resp, outErr = callback(c, fCtx, req)
		return true
	})
	if err != nil {
		return nil, err
	}
	return resp, outErr
}

// forEachNode executes callback for each node in the container until true is returned.
// Returns errNoSuitableNode if there was no successful attempt to dial any node.
func (s *Service) forEachNode(ctx context.Context, cntNodes []netmapSDK.NodeInfo, f func(context.Context, TreeServiceClient) bool) error {
	for _, n := range cntNodes {
		if bytes.Equal(n.PublicKey(), s.rawPub) {
			return nil
		}
	}

	var called bool
	for _, n := range cntNodes {
		var stop bool
		for endpoint := range n.NetworkEndpoints() {
			stop = s.execOnClient(ctx, endpoint, func(fCtx context.Context, c TreeServiceClient) bool {
				called = true
				return f(fCtx, c)
			})
			if called {
				break
			}
		}
		if stop {
			return nil
		}
	}
	if !called {
		return errNoSuitableNode
	}
	return nil
}

func (s *Service) execOnClient(ctx context.Context, endpoint string, f func(context.Context, TreeServiceClient) bool) bool {
	ctx, span := tracing.StartSpanFromContext(ctx, "TreeService.IterateNetworkEndpoints",
		trace.WithAttributes(
			attribute.String("endpoint", endpoint),
		))
	defer span.End()

	c, err := s.cache.get(ctx, endpoint)
	if err != nil {
		return false
	}

	s.log.Debug(ctx, logs.TreeRedirectingTreeServiceQuery, zap.String("endpoint", endpoint))
	return f(ctx, c)
}