[#488] reputation/remoteProvider: Implement intermediate remote

Move common remoteProvider code to cmd/reputation/common.
Hide WriterProvider initialization behind interface and
add implementation of that interface to local and
intermediate packages in cmd/reputation directory.

Signed-off-by: Pavel Karpy <carpawell@nspcc.ru>
This commit is contained in:
Pavel Karpy 2021-04-23 17:49:23 +03:00 committed by Alex Vanin
parent 49d477f466
commit e8885d72f4
4 changed files with 235 additions and 64 deletions

View file

@ -0,0 +1,90 @@
package common
import (
apiClient "github.com/nspcc-dev/neofs-api-go/pkg/client"
"github.com/nspcc-dev/neofs-node/pkg/network"
reputationcommon "github.com/nspcc-dev/neofs-node/pkg/services/reputation/common"
reputationrouter "github.com/nspcc-dev/neofs-node/pkg/services/reputation/common/router"
trustcontroller "github.com/nspcc-dev/neofs-node/pkg/services/reputation/local/controller"
"github.com/pkg/errors"
)
type clientCache interface {
Get(string) (apiClient.Client, error)
}
// clientKeyRemoteProvider must provide remote writer and take into account
// that requests must be sent via passed api client and must be signed with
// passed private key.
type clientKeyRemoteProvider interface {
WithClient(apiClient.Client) reputationcommon.WriterProvider
}
// remoteTrustProvider is implementation of reputation RemoteWriterProvider interface.
// It caches clients, checks if it is the end of the route and checks either current
// node is remote target or not.
//
// remoteTrustProvider requires to be provided with clientKeyRemoteProvider.
type remoteTrustProvider struct {
localAddrSrc network.LocalAddressSource
deadEndProvider reputationcommon.WriterProvider
clientCache clientCache
remoteProvider clientKeyRemoteProvider
}
// RemoteProviderPrm groups the required parameters of the remoteTrustProvider's constructor.
//
// All values must comply with the requirements imposed on them.
// Passing incorrect parameter values will result in constructor
// failure (error or panic depending on the implementation).
type RemoteProviderPrm struct {
LocalAddrSrc network.LocalAddressSource
DeadEndProvider reputationcommon.WriterProvider
ClientCache clientCache
WriterProvider clientKeyRemoteProvider
}
func NewRemoteTrustProvider(prm RemoteProviderPrm) reputationrouter.RemoteWriterProvider {
switch {
case prm.LocalAddrSrc == nil:
PanicOnPrmValue("LocalAddrSrc", prm.LocalAddrSrc)
case prm.DeadEndProvider == nil:
PanicOnPrmValue("DeadEndProvider", prm.DeadEndProvider)
case prm.ClientCache == nil:
PanicOnPrmValue("ClientCache", prm.ClientCache)
case prm.WriterProvider == nil:
PanicOnPrmValue("WriterProvider", prm.WriterProvider)
}
return &remoteTrustProvider{
localAddrSrc: prm.LocalAddrSrc,
deadEndProvider: prm.DeadEndProvider,
clientCache: prm.ClientCache,
remoteProvider: prm.WriterProvider,
}
}
func (rtp *remoteTrustProvider) InitRemote(srv reputationcommon.ServerInfo) (reputationcommon.WriterProvider, error) {
if srv == nil {
return rtp.deadEndProvider, nil
}
addr := srv.Address()
if rtp.localAddrSrc.LocalAddress().String() == srv.Address() {
// if local => return no-op writer
return trustcontroller.SimpleWriterProvider(new(NopReputationWriter)), nil
}
ipAddr, err := network.IPAddrFromMultiaddr(addr)
if err != nil {
return nil, errors.Wrap(err, "could not convert address to IP format")
}
c, err := rtp.clientCache.Get(ipAddr)
if err != nil {
return nil, errors.Wrap(err, "could not initialize API client")
}
return rtp.remoteProvider.WithClient(c), nil
}

View file

@ -0,0 +1,120 @@
package intermediate
import (
"crypto/ecdsa"
apiClient "github.com/nspcc-dev/neofs-api-go/pkg/client"
reputationapi "github.com/nspcc-dev/neofs-api-go/pkg/reputation"
"github.com/nspcc-dev/neofs-node/cmd/neofs-node/reputation/common"
"github.com/nspcc-dev/neofs-node/pkg/services/reputation"
reputationcommon "github.com/nspcc-dev/neofs-node/pkg/services/reputation/common"
eigentrustcalc "github.com/nspcc-dev/neofs-node/pkg/services/reputation/eigentrust/calculator"
)
// RemoteProviderPrm groups the required parameters of the RemoteProvider's constructor.
//
// All values must comply with the requirements imposed on them.
// Passing incorrect parameter values will result in constructor
// failure (error or panic depending on the implementation).
type RemoteProviderPrm struct {
Key *ecdsa.PrivateKey
}
// NewRemoteProvider creates a new instance of the RemoteProvider.
//
// Panics if at least one value of the parameters is invalid.
//
// The created RemoteProvider does not require additional
// initialization and is completely ready for work.
func NewRemoteProvider(prm RemoteProviderPrm) *RemoteProvider {
switch {
case prm.Key == nil:
common.PanicOnPrmValue("NetMapSource", prm.Key)
}
return &RemoteProvider{
key: prm.Key,
}
}
// RemoteProvider is an implementation of the clientKeyRemoteProvider interface.
type RemoteProvider struct {
key *ecdsa.PrivateKey
}
func (rp RemoteProvider) WithClient(c apiClient.Client) reputationcommon.WriterProvider {
return &TrustWriterProvider{
client: c,
key: rp.key,
}
}
type TrustWriterProvider struct {
client apiClient.Client
key *ecdsa.PrivateKey
}
func (twp *TrustWriterProvider) InitWriter(ctx reputationcommon.Context) (reputationcommon.Writer, error) {
return &RemoteTrustWriter{
ctx: ctx,
client: twp.client,
key: twp.key,
}, nil
}
type RemoteTrustWriter struct {
ctx reputationcommon.Context
client apiClient.Client
key *ecdsa.PrivateKey
buf []*apiClient.SendIntermediateTrustPrm
}
// Write check if passed context contains required
// data(returns ErrIncorrectContext if not) and
// caches passed trusts(as SendIntermediateTrustPrm structs).
func (rtp *RemoteTrustWriter) Write(ctx reputationcommon.Context, t reputation.Trust) error {
eiContext, ok := ctx.(eigentrustcalc.Context)
if !ok {
return ErrIncorrectContext
}
apiTrustingPeer := reputationapi.NewPeerID()
apiTrustingPeer.SetPublicKey(t.TrustingPeer())
apiTrustedPeer := reputationapi.NewPeerID()
apiTrustedPeer.SetPublicKey(t.Peer())
apiTrust := reputationapi.NewTrust()
apiTrust.SetValue(t.Value().Float64())
apiTrust.SetPeer(apiTrustedPeer)
apiPeerToPeerTrust := reputationapi.NewPeerToPeerTrust()
apiPeerToPeerTrust.SetTrustingPeer(apiTrustingPeer)
apiPeerToPeerTrust.SetTrust(apiTrust)
p := &apiClient.SendIntermediateTrustPrm{}
p.SetEpoch(eiContext.Epoch())
p.SetIteration(eiContext.I())
p.SetTrust(apiPeerToPeerTrust)
rtp.buf = append(rtp.buf, p)
return nil
}
// Close sends all cached intermediate trusts.
// If error occurs, returns in immediately and stops iteration.
func (rtp *RemoteTrustWriter) Close() (err error) {
for _, prm := range rtp.buf {
_, err = rtp.client.SendIntermediateTrust(
rtp.ctx,
*prm,
apiClient.WithKey(rtp.key),
)
if err != nil {
return
}
}
return
}

View file

@ -5,98 +5,59 @@ import (
apiClient "github.com/nspcc-dev/neofs-api-go/pkg/client" apiClient "github.com/nspcc-dev/neofs-api-go/pkg/client"
reputationapi "github.com/nspcc-dev/neofs-api-go/pkg/reputation" reputationapi "github.com/nspcc-dev/neofs-api-go/pkg/reputation"
reputationutil "github.com/nspcc-dev/neofs-node/cmd/neofs-node/reputation/common" "github.com/nspcc-dev/neofs-node/cmd/neofs-node/reputation/common"
"github.com/nspcc-dev/neofs-node/pkg/network"
"github.com/nspcc-dev/neofs-node/pkg/services/reputation" "github.com/nspcc-dev/neofs-node/pkg/services/reputation"
reputationcommon "github.com/nspcc-dev/neofs-node/pkg/services/reputation/common" reputationcommon "github.com/nspcc-dev/neofs-node/pkg/services/reputation/common"
reputationrouter "github.com/nspcc-dev/neofs-node/pkg/services/reputation/common/router"
trustcontroller "github.com/nspcc-dev/neofs-node/pkg/services/reputation/local/controller"
"github.com/pkg/errors"
) )
type clientCache interface { // RemoteProviderPrm groups the required parameters of the RemoteProvider's constructor.
Get(string) (apiClient.Client, error)
}
// remoteTrustProvider is implementation of reputation RemoteWriterProvider interface.
type remoteTrustProvider struct {
localAddrSrc network.LocalAddressSource
deadEndProvider reputationcommon.WriterProvider
key *ecdsa.PrivateKey
clientCache clientCache
}
// RemoteProviderPrm groups the required parameters of the remoteTrustProvider's constructor.
// //
// All values must comply with the requirements imposed on them. // All values must comply with the requirements imposed on them.
// Passing incorrect parameter values will result in constructor // Passing incorrect parameter values will result in constructor
// failure (error or panic depending on the implementation). // failure (error or panic depending on the implementation).
type RemoteProviderPrm struct { type RemoteProviderPrm struct {
LocalAddrSrc network.LocalAddressSource Key *ecdsa.PrivateKey
DeadEndProvider reputationcommon.WriterProvider
Key *ecdsa.PrivateKey
ClientCache clientCache
} }
func NewRemoteTrustProvider(prm RemoteProviderPrm) reputationrouter.RemoteWriterProvider { // NewRemoteProvider creates a new instance of the RemoteProvider.
//
// Panics if at least one value of the parameters is invalid.
//
// The created RemoteProvider does not require additional
// initialization and is completely ready for work.
func NewRemoteProvider(prm RemoteProviderPrm) *RemoteProvider {
switch { switch {
case prm.LocalAddrSrc == nil:
reputationutil.PanicOnPrmValue("LocalAddrSrc", prm.LocalAddrSrc)
case prm.DeadEndProvider == nil:
reputationutil.PanicOnPrmValue("DeadEndProvider", prm.DeadEndProvider)
case prm.Key == nil: case prm.Key == nil:
reputationutil.PanicOnPrmValue("Key", prm.Key) common.PanicOnPrmValue("NetMapSource", prm.Key)
case prm.ClientCache == nil:
reputationutil.PanicOnPrmValue("ClientCache", prm.ClientCache)
} }
return &remoteTrustProvider{ return &RemoteProvider{
localAddrSrc: prm.LocalAddrSrc, key: prm.Key,
deadEndProvider: prm.DeadEndProvider,
key: prm.Key,
clientCache: prm.ClientCache,
} }
} }
func (rtp *remoteTrustProvider) InitRemote(srv reputationrouter.ServerInfo) (reputationcommon.WriterProvider, error) { // RemoteProvider is an implementation of the clientKeyRemoteProvider interface.
if srv == nil { type RemoteProvider struct {
return rtp.deadEndProvider, nil key *ecdsa.PrivateKey
} }
addr := srv.Address() func (rp RemoteProvider) WithClient(c apiClient.Client) reputationcommon.WriterProvider {
return &TrustWriterProvider{
if rtp.localAddrSrc.LocalAddress().String() == srv.Address() {
// if local => return no-op writer
return trustcontroller.SimpleWriterProvider(new(reputationutil.NopReputationWriter)), nil
}
ipAddr, err := network.IPAddrFromMultiaddr(addr)
if err != nil {
return nil, errors.Wrap(err, "could not convert address to IP format")
}
c, err := rtp.clientCache.Get(ipAddr)
if err != nil {
return nil, errors.Wrap(err, "could not initialize API client")
}
return &RemoteTrustWriterProvider{
client: c, client: c,
key: rtp.key, key: rp.key,
}, nil }
} }
type RemoteTrustWriterProvider struct { type TrustWriterProvider struct {
client apiClient.Client client apiClient.Client
key *ecdsa.PrivateKey key *ecdsa.PrivateKey
} }
func (rtwp *RemoteTrustWriterProvider) InitWriter(ctx reputationcommon.Context) (reputationcommon.Writer, error) { func (twp *TrustWriterProvider) InitWriter(ctx reputationcommon.Context) (reputationcommon.Writer, error) {
return &RemoteTrustWriter{ return &RemoteTrustWriter{
ctx: ctx, ctx: ctx,
client: rtwp.client, client: twp.client,
key: rtwp.key, key: twp.key,
}, nil }, nil
} }

BIN
go.sum

Binary file not shown.