frostfs-node/cmd/neofs-node/reputation.go

package main

import (
	"context"

	v2reputation "github.com/nspcc-dev/neofs-api-go/v2/reputation"
	v2reputationgrpc "github.com/nspcc-dev/neofs-api-go/v2/reputation/grpc"
	crypto "github.com/nspcc-dev/neofs-crypto"
	"github.com/nspcc-dev/neofs-node/cmd/neofs-node/reputation/intermediate"
	localreputation "github.com/nspcc-dev/neofs-node/cmd/neofs-node/reputation/local"
	"github.com/nspcc-dev/neofs-node/pkg/morph/event"
	"github.com/nspcc-dev/neofs-node/pkg/morph/event/netmap"
	"github.com/nspcc-dev/neofs-node/pkg/network/cache"
	grpcreputation "github.com/nspcc-dev/neofs-node/pkg/network/transport/reputation/grpc"
	"github.com/nspcc-dev/neofs-node/pkg/services/reputation"
	reputationcommon "github.com/nspcc-dev/neofs-node/pkg/services/reputation/common"
	reputationrouter "github.com/nspcc-dev/neofs-node/pkg/services/reputation/common/router"
	"github.com/nspcc-dev/neofs-node/pkg/services/reputation/eigentrust/storage/daughters"
	trustcontroller "github.com/nspcc-dev/neofs-node/pkg/services/reputation/local/controller"
	"github.com/nspcc-dev/neofs-node/pkg/services/reputation/local/managers"
	truststorage "github.com/nspcc-dev/neofs-node/pkg/services/reputation/local/storage"
	reputationrpc "github.com/nspcc-dev/neofs-node/pkg/services/reputation/rpc"
	"github.com/nspcc-dev/neofs-node/pkg/util/logger"
	"github.com/pkg/errors"
	"go.uber.org/zap"
)

func initReputationService(c *cfg) {
	// consider sharing this between application components
	nmSrc := newCachedNetmapStorage(c.cfgNetmap.state, c.cfgNetmap.wrapper)

	// storing calculated trusts as a daughter
	c.cfgReputation.localTrustStorage = truststorage.New(truststorage.Prm{})

	// storing received trusts as a manager
	daughterStorage := &intermediate.DaughterStorage{
		Log:     c.log,
		Storage: daughters.New(daughters.Prm{}),
	}

	trustStorage := &localreputation.TrustStorage{
		Log:      c.log,
		Storage:  c.cfgReputation.localTrustStorage,
		NmSrc:    nmSrc,
		LocalKey: crypto.MarshalPublicKey(&c.key.PublicKey),
	}

	managerBuilder := localreputation.NewManagerBuilder(
		localreputation.ManagersPrm{
			NetMapSource: nmSrc,
		},
		localreputation.WithLogger(c.log),
	)

	routeBuilder := managers.New(managers.Prm{
		ManagerBuilder: managerBuilder,
	})

	remoteLocalTrustProvider := localreputation.NewRemoteTrustProvider(
		localreputation.RemoteProviderPrm{
			LocalAddrSrc:    c,
			DeadEndProvider: daughterStorage,
			ClientCache:     cache.NewSDKClientCache(),
			Key:             c.key,
		},
	)

	router := reputationrouter.New(
		reputationrouter.Prm{
			LocalServerInfo:      c,
			RemoteWriterProvider: remoteLocalTrustProvider,
			Builder:              routeBuilder,
		})

	c.cfgReputation.localTrustCtrl = trustcontroller.New(trustcontroller.Prm{
		LocalTrustSource: trustStorage,
		LocalTrustTarget: router,
	})

	addNewEpochAsyncNotificationHandler(
		c,
		func(ev event.Event) {
			var reportPrm trustcontroller.ReportPrm

			// report collected values from previous epoch
			reportPrm.SetEpoch(ev.(netmap.NewEpoch).EpochNumber() - 1)

			c.cfgReputation.localTrustCtrl.Report(reportPrm)
		},
	)

	v2reputationgrpc.RegisterReputationServiceServer(c.cfgGRPC.server,
		grpcreputation.New(
			reputationrpc.NewSignService(
				c.key,
				reputationrpc.NewResponseService(
					&reputationServer{
						cfg:          c,
						log:          c.log,
						router:       router,
						routeBuilder: routeBuilder,
					},
					c.respSvc,
				),
			),
		),
	)

	// initialize eigen trust block timer
	durationMeter := NewEigenTrustDuration(c.cfgNetmap.wrapper)

	newEigenTrustIterTimer(c, durationMeter, func() {
		c.log.Debug("todo: start next EigenTrust iteration round")
	})

	addNewEpochAsyncNotificationHandler(
		c,
		func(e event.Event) {
			durationMeter.Update() // recalculate duration of one iteration round

			err := c.cfgMorph.eigenTrustTimer.Reset() // start iteration rounds again
			if err != nil {
				c.log.Warn("can't reset block timer to start eigen trust calculations again",
					zap.String("error", err.Error()))
			}
		},
	)
}

type reputationServer struct {
	*cfg
	log          *logger.Logger
	router       reputationcommon.WriterProvider
	routeBuilder reputationrouter.Builder
}

func (s *reputationServer) SendLocalTrust(ctx context.Context, req *v2reputation.SendLocalTrustRequest) (*v2reputation.SendLocalTrustResponse, error) {
	var passedRoute []reputationrouter.ServerInfo

	for hdr := req.GetVerificationHeader(); hdr != nil; hdr = hdr.GetOrigin() {
		passedRoute = append(passedRoute, &localreputation.OnlyKeyRemoteServerInfo{
			Key: hdr.GetBodySignature().GetKey(),
		})
	}

	for left, right := 0, len(passedRoute)-1; left < right; left, right = left+1, right-1 {
		passedRoute[left], passedRoute[right] = passedRoute[right], passedRoute[left]
	}

	passedRoute = append(passedRoute, s)

	body := req.GetBody()

	eCtx := &localreputation.EpochContext{
		Context: ctx,
		E:       body.GetEpoch(),
	}

	w, err := s.router.InitWriter(reputationrouter.NewRouteContext(eCtx, passedRoute))
	if err != nil {
		return nil, errors.Wrap(err, "could not initialize local trust writer")
	}

	for _, trust := range body.GetTrusts() {
		err = s.processTrust(body.GetEpoch(), apiToLocalTrust(trust, passedRoute[0].PublicKey()), passedRoute, w)
		if err != nil {
			return nil, errors.Wrap(err, "could not write one of trusts")
		}
	}

	resp := new(v2reputation.SendLocalTrustResponse)
	resp.SetBody(new(v2reputation.SendLocalTrustResponseBody))

	return resp, nil
}

func (s *reputationServer) SendIntermediateResult(_ context.Context, req *v2reputation.SendIntermediateResultRequest) (*v2reputation.SendIntermediateResultResponse, error) {
	resp := new(v2reputation.SendIntermediateResultResponse)

	// todo: implement me

	return resp, nil
}

// apiToLocalTrust converts v2 Trust to local reputation.Trust, adding trustingPeer.
func apiToLocalTrust(t *v2reputation.Trust, trustingPeer []byte) reputation.Trust {
	localTrust := reputation.Trust{}

	localTrust.SetValue(reputation.TrustValueFromFloat64(t.GetValue()))
	localTrust.SetPeer(reputation.PeerIDFromBytes(t.GetPeer().GetValue()))
	localTrust.SetTrustingPeer(reputation.PeerIDFromBytes(trustingPeer))

	return localTrust
}

func (s *reputationServer) processTrust(epoch uint64, t reputation.Trust,
	passedRoute []reputationrouter.ServerInfo, w reputationcommon.Writer) error {
	err := reputationrouter.CheckRoute(s.routeBuilder, epoch, t, passedRoute)
	if err != nil {
		return errors.Wrap(err, "wrong route of reputation trust value")
	}

	return w.Write(&localreputation.EpochContext{E: epoch}, t)
}