TrueCloudLab/frostfs-node
20
1
Fork 27
Code Issues 95 Pull requests 4 Projects Releases 1 Activity Actions

[#360] Share common parts of basic and audit settlements

Browse source 
Signed-off-by: Alex Vanin <alexey@nspcc.ru>
This commit is contained in:
Alex Vanin 2021-02-01 15:40:07 +03:00 committed by Alex Vanin
parent bd5c70131b
commit f45675b7a2
7 changed files with 180 additions and 158 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
pkg/innerring/innerring.go
View file

@ -284,7 +284,7 @@ func New(ctx context.Context, log *zap.Logger, cfg *viper.Viper) (*Server, error
}
// create settlement processor dependencies
auditCalcDeps := &auditSettlementDeps{
settlementDeps := &settlementDeps{
log: server.log,
cnrSrc: cnrClient,
auditClient: server.auditClient,
@ -293,6 +293,10 @@ func New(ctx context.Context, log *zap.Logger, cfg *viper.Viper) (*Server, error
balanceClient: balClient,
}
auditCalcDeps := &auditSettlementDeps{
settlementDeps: settlementDeps,
}
auditSettlementCalc := auditSettlement.NewCalculator(
&auditSettlement.CalculatorPrm{
ResultStorage: auditCalcDeps,

36
pkg/innerring/processors/settlement/audit/calculate.go
View file

@ -8,6 +8,7 @@ import (
"github.com/nspcc-dev/neofs-api-go/pkg/audit"
"github.com/nspcc-dev/neofs-api-go/pkg/container"
"github.com/nspcc-dev/neofs-api-go/pkg/object"
"github.com/nspcc-dev/neofs-node/pkg/innerring/processors/settlement/common"
"github.com/nspcc-dev/neofs-node/pkg/util/logger"
"go.uber.org/zap"
)
@ -28,13 +29,13 @@ type singleResultCtx struct {
cid *container.ID
txTable *transferTable
txTable *common.TransferTable
cnrInfo ContainerInfo
cnrInfo common.ContainerInfo
cnrNodes []NodeInfo
cnrNodes []common.NodeInfo
passNodes map[string]NodeInfo
passNodes map[string]common.NodeInfo
sumSGSize *big.Int
}
@ -69,7 +70,7 @@ func (c *Calculator) Calculate(p *CalculatePrm) {
zap.Int("number", len(auditResults)),
)
table := newTransferTable()
table := common.NewTransferTable()
for i := range auditResults {
c.processResult(&singleResultCtx{
@ -81,20 +82,7 @@ func (c *Calculator) Calculate(p *CalculatePrm) {
log.Debug("processing transfers")
table.iterate(func(tx *transferTx) {
sign := tx.amount.Sign()
if sign == 0 {
log.Debug("ignore zero transfer")
return
}
if sign < 0 {
tx.from, tx.to = tx.to, tx.from
tx.amount.Neg(tx.amount)
}
c.prm.Exchanger.Transfer(tx.from, tx.to, tx.amount)
})
common.TransferAssets(c.prm.Exchanger, table)
}
func (c *Calculator) processResult(ctx *singleResultCtx) {
@ -168,7 +156,7 @@ func (c *Calculator) buildPlacement(ctx *singleResultCtx) bool {
}
func (c *Calculator) collectPassNodes(ctx *singleResultCtx) bool {
ctx.passNodes = make(map[string]NodeInfo)
ctx.passNodes = make(map[string]common.NodeInfo)
loop:
for _, cnrNode := range ctx.cnrNodes {
@ -261,10 +249,10 @@ func (c *Calculator) fillTransferTable(ctx *singleResultCtx) bool {
fee.Add(fee, bigOne)
}
ctx.txTable.transfer(&transferTx{
from: cnrOwner,
to: ownerID,
amount: fee,
ctx.txTable.Transfer(&common.TransferTx{
From: cnrOwner,
To: ownerID,
Amount: fee,
})
}

59
pkg/innerring/processors/settlement/audit/prm.go
View file

@ -1,27 +1,24 @@
package audit
import (
"math/big"
"github.com/nspcc-dev/neofs-api-go/pkg/audit"
"github.com/nspcc-dev/neofs-api-go/pkg/container"
"github.com/nspcc-dev/neofs-api-go/pkg/object"
"github.com/nspcc-dev/neofs-api-go/pkg/owner"
"github.com/nspcc-dev/neofs-node/pkg/innerring/processors/settlement/common"
)
// CalculatorPrm groups the parameters of Calculator's constructor.
type CalculatorPrm struct {
ResultStorage ResultStorage
ContainerStorage ContainerStorage
ContainerStorage common.ContainerStorage
PlacementCalculator PlacementCalculator
PlacementCalculator common.PlacementCalculator
SGStorage SGStorage
AccountStorage AccountStorage
AccountStorage common.AccountStorage
Exchanger Exchanger
Exchanger common.Exchanger
}
// ResultStorage is an interface of storage of the audit results.
@ -30,37 +27,6 @@ type ResultStorage interface {
AuditResultsForEpoch(epoch uint64) ([]*audit.Result, error)
}
// NodeInfo groups the data about the storage node
// necessary for calculating audit fees.
type NodeInfo interface {
// Must return storage price of the node for one epoch in GASe-12.
Price() *big.Int
// Must return public key of the node.
PublicKey() []byte
}
// ContainerInfo groups the data about NeoFS container
// necessary for calculating audit fee.
type ContainerInfo interface {
// Must return identifier of the container owner.
Owner() *owner.ID
}
// ContainerStorage is an interface of
// storage of the NeoFS containers.
type ContainerStorage interface {
// Must return information about the container by ID.
ContainerInfo(*container.ID) (ContainerInfo, error)
}
// PlacementCalculator is a component interface
// that builds placement vectors.
type PlacementCalculator interface {
// Must return information about the nodes from container cid of the epoch e.
ContainerNodes(e uint64, cid *container.ID) ([]NodeInfo, error)
}
// SGInfo groups the data about NeoFS storage group
// necessary for calculating audit fee.
type SGInfo interface {
@ -73,18 +39,3 @@ type SGStorage interface {
// Must return information about the storage group by address.
SGInfo(*object.Address) (SGInfo, error)
}
// AccountStorage is an network member accounts interface.
type AccountStorage interface {
// Must resolve information about the storage node
// to its ID in system.
ResolveKey(NodeInfo) (*owner.ID, error)
}
// Exchanger is an interface of monetary component.
type Exchanger interface {
// Must transfer amount of GASe-12 from sender to recipient.
//
// Amount must be positive.
Transfer(sender, recipient *owner.ID, amount *big.Int)
}

57
pkg/innerring/processors/settlement/audit/util.go
View file

@ -1,57 +0,0 @@
package audit
import (
"math/big"
"github.com/nspcc-dev/neofs-api-go/pkg/owner"
)
type transferTable struct {
txs map[string]map[string]*transferTx
}
type transferTx struct {
from, to *owner.ID
amount *big.Int
}
func newTransferTable() *transferTable {
return &transferTable{
txs: make(map[string]map[string]*transferTx),
}
}
func (t *transferTable) transfer(tx *transferTx) {
from, to := tx.from.String(), tx.to.String()
if from == to {
return
}
m, ok := t.txs[from]
if !ok {
if m, ok = t.txs[to]; ok {
to = from // ignore `from = to` swap because `from` doesn't require
tx.amount.Neg(tx.amount)
} else {
m = make(map[string]*transferTx, 1)
t.txs[from] = m
}
}
tgt, ok := m[to]
if !ok {
m[to] = tx
return
}
tgt.amount.Add(tgt.amount, tx.amount)
}
func (t *transferTable) iterate(f func(*transferTx)) {
for _, m := range t.txs {
for _, tx := range m {
f(tx)
}
}
}

54
pkg/innerring/processors/settlement/common/types.go Normal file
View file

@ -0,0 +1,54 @@
package common
import (
"math/big"
"github.com/nspcc-dev/neofs-api-go/pkg/container"
"github.com/nspcc-dev/neofs-api-go/pkg/owner"
)
// NodeInfo groups the data about the storage node
// necessary for calculating audit fees.
type NodeInfo interface {
// Must return storage price of the node for one epoch in GASe-12.
Price() *big.Int
// Must return public key of the node.
PublicKey() []byte
}
// ContainerInfo groups the data about NeoFS container
// necessary for calculating audit fee.
type ContainerInfo interface {
// Must return identifier of the container owner.
Owner() *owner.ID
}
// ContainerStorage is an interface of
// storage of the NeoFS containers.
type ContainerStorage interface {
// Must return information about the container by ID.
ContainerInfo(*container.ID) (ContainerInfo, error)
}
// PlacementCalculator is a component interface
// that builds placement vectors.
type PlacementCalculator interface {
// Must return information about the nodes from container cid of the epoch e.
ContainerNodes(e uint64, cid *container.ID) ([]NodeInfo, error)
}
// AccountStorage is an network member accounts interface.
type AccountStorage interface {
// Must resolve information about the storage node
// to its ID in system.
ResolveKey(NodeInfo) (*owner.ID, error)
}
// Exchanger is an interface of monetary component.
type Exchanger interface {
// Must transfer amount of GASe-12 from sender to recipient.
//
// Amount must be positive.
Transfer(sender, recipient *owner.ID, amount *big.Int)
}

73
pkg/innerring/processors/settlement/common/util.go Normal file
View file

@ -0,0 +1,73 @@
package common
import (
"math/big"
"github.com/nspcc-dev/neofs-api-go/pkg/owner"
)
type TransferTable struct {
txs map[string]map[string]*TransferTx
}
type TransferTx struct {
From, To *owner.ID
Amount *big.Int
}
func NewTransferTable() *TransferTable {
return &TransferTable{
txs: make(map[string]map[string]*TransferTx),
}
}
func (t *TransferTable) Transfer(tx *TransferTx) {
from, to := tx.From.String(), tx.To.String()
if from == to {
return
}
m, ok := t.txs[from]
if !ok {
if m, ok = t.txs[to]; ok {
to = from // ignore `From = To` swap because `From` doesn't require
tx.Amount.Neg(tx.Amount)
} else {
m = make(map[string]*TransferTx, 1)
t.txs[from] = m
}
}
tgt, ok := m[to]
if !ok {
m[to] = tx
return
}
tgt.Amount.Add(tgt.Amount, tx.Amount)
}
func (t *TransferTable) Iterate(f func(*TransferTx)) {
for _, m := range t.txs {
for _, tx := range m {
f(tx)
}
}
}
func TransferAssets(e Exchanger, t *TransferTable) {
t.Iterate(func(tx *TransferTx) {
sign := tx.Amount.Sign()
if sign == 0 {
return
}
if sign < 0 {
tx.From, tx.To = tx.To, tx.From
tx.Amount.Neg(tx.Amount)
}
e.Transfer(tx.From, tx.To, tx.Amount)
})
}

53
pkg/innerring/audit.go → pkg/innerring/settlement.go
View file

@ -14,6 +14,7 @@ import (
"github.com/nspcc-dev/neofs-node/pkg/core/container"
"github.com/nspcc-dev/neofs-node/pkg/core/netmap"
"github.com/nspcc-dev/neofs-node/pkg/innerring/processors/settlement/audit"
"github.com/nspcc-dev/neofs-node/pkg/innerring/processors/settlement/common"
auditClient "github.com/nspcc-dev/neofs-node/pkg/morph/client/audit/wrapper"
balanceClient "github.com/nspcc-dev/neofs-node/pkg/morph/client/balance/wrapper"
"github.com/nspcc-dev/neofs-node/pkg/util/logger"
@ -21,7 +22,7 @@ import (
"go.uber.org/zap"
)
type auditSettlementDeps struct {
type settlementDeps struct {
log *logger.Logger
cnrSrc container.Source
@ -35,6 +36,10 @@ type auditSettlementDeps struct {
balanceClient *balanceClient.Wrapper
}
type auditSettlementDeps struct {
*settlementDeps
}
type auditSettlementCalculator audit.Calculator
type containerWrapper containerAPI.Container
@ -61,8 +66,8 @@ func (c *containerWrapper) Owner() *owner.ID {
return (*containerAPI.Container)(c).OwnerID()
}
func (a auditSettlementDeps) AuditResultsForEpoch(epoch uint64) ([]*auditAPI.Result, error) {
idList, err := a.auditClient.ListAuditResultIDByEpoch(epoch)
func (s settlementDeps) AuditResultsForEpoch(epoch uint64) ([]*auditAPI.Result, error) {
idList, err := s.auditClient.ListAuditResultIDByEpoch(epoch)
if err != nil {
return nil, errors.Wrap(err, "could not list audit results in sidechain")
}
@ -70,7 +75,7 @@ func (a auditSettlementDeps) AuditResultsForEpoch(epoch uint64) ([]*auditAPI.Res
res := make([]*auditAPI.Result, 0, len(idList))
for i := range idList {
r, err := a.auditClient.GetAuditResult(idList[i])
r, err := s.auditClient.GetAuditResult(idList[i])
if err != nil {
return nil, errors.Wrap(err, "could not get audit result")
}
@ -81,8 +86,8 @@ func (a auditSettlementDeps) AuditResultsForEpoch(epoch uint64) ([]*auditAPI.Res
return res, nil
}
func (a auditSettlementDeps) ContainerInfo(cid *containerAPI.ID) (audit.ContainerInfo, error) {
cnr, err := a.cnrSrc.Get(cid)
func (s settlementDeps) ContainerInfo(cid *containerAPI.ID) (common.ContainerInfo, error) {
cnr, err := s.cnrSrc.Get(cid)
if err != nil {
return nil, errors.Wrap(err, "could not get container from storage")
}
@ -90,23 +95,23 @@ func (a auditSettlementDeps) ContainerInfo(cid *containerAPI.ID) (audit.Containe
return (*containerWrapper)(cnr), nil
}
func (a auditSettlementDeps) buildContainer(e uint64, cid *containerAPI.ID) (netmapAPI.ContainerNodes, *netmapAPI.Netmap, error) {
func (s settlementDeps) buildContainer(e uint64, cid *containerAPI.ID) (netmapAPI.ContainerNodes, *netmapAPI.Netmap, error) {
var (
nm *netmapAPI.Netmap
err error
)
if e > 0 {
nm, err = a.nmSrc.GetNetMapByEpoch(e)
nm, err = s.nmSrc.GetNetMapByEpoch(e)
} else {
nm, err = netmap.GetLatestNetworkMap(a.nmSrc)
nm, err = netmap.GetLatestNetworkMap(s.nmSrc)
}
if err != nil {
return nil, nil, errors.Wrap(err, "could not get network map from storage")
}
cnr, err := a.cnrSrc.Get(cid)
cnr, err := s.cnrSrc.Get(cid)
if err != nil {
return nil, nil, errors.Wrap(err, "could not get container from sidechain")
}
@ -122,14 +127,14 @@ func (a auditSettlementDeps) buildContainer(e uint64, cid *containerAPI.ID) (net
return cn, nm, nil
}
func (a auditSettlementDeps) ContainerNodes(e uint64, cid *containerAPI.ID) ([]audit.NodeInfo, error) {
cn, _, err := a.buildContainer(e, cid)
func (s settlementDeps) ContainerNodes(e uint64, cid *containerAPI.ID) ([]common.NodeInfo, error) {
cn, _, err := s.buildContainer(e, cid)
if err != nil {
return nil, err
}
ns := cn.Flatten()
res := make([]audit.NodeInfo, 0, len(ns))
res := make([]common.NodeInfo, 0, len(ns))
for i := range ns {
res = append(res, &nodeInfoWrapper{
@ -140,13 +145,13 @@ func (a auditSettlementDeps) ContainerNodes(e uint64, cid *containerAPI.ID) ([]a
return res, nil
}
func (a auditSettlementDeps) SGInfo(addr *object.Address) (audit.SGInfo, error) {
cn, nm, err := a.buildContainer(0, addr.ContainerID())
func (s settlementDeps) SGInfo(addr *object.Address) (audit.SGInfo, error) {
cn, nm, err := s.buildContainer(0, addr.ContainerID())
if err != nil {
return nil, err
}
sg, err := a.clientCache.getSG(context.Background(), addr, nm, cn)
sg, err := s.clientCache.getSG(context.Background(), addr, nm, cn)
if err != nil {
return nil, err
}
@ -154,7 +159,7 @@ func (a auditSettlementDeps) SGInfo(addr *object.Address) (audit.SGInfo, error)
return (*sgWrapper)(sg), nil
}
func (a auditSettlementDeps) ResolveKey(ni audit.NodeInfo) (*owner.ID, error) {
func (s settlementDeps) ResolveKey(ni common.NodeInfo) (*owner.ID, error) {
w, err := owner.NEO3WalletFromPublicKey(crypto.UnmarshalPublicKey(ni.PublicKey()))
if err != nil {
return nil, err
@ -168,24 +173,24 @@ func (a auditSettlementDeps) ResolveKey(ni audit.NodeInfo) (*owner.ID, error) {
var transferAuditDetails = []byte("settlement-audit")
func (a auditSettlementDeps) Transfer(sender, recipient *owner.ID, amount *big.Int) {
log := a.log.With(
func (s settlementDeps) transfer(sender, recipient *owner.ID, amount *big.Int, details []byte) {
log := s.log.With(
zap.Stringer("sender", sender),
zap.Stringer("recipient", recipient),
zap.Stringer("amount (GASe-12)", amount),
)
if !amount.IsInt64() {
a.log.Error("amount can not be represented as an int64")
s.log.Error("amount can not be represented as an int64")
return
}
if err := a.balanceClient.TransferX(balanceClient.TransferPrm{
if err := s.balanceClient.TransferX(balanceClient.TransferPrm{
Amount: amount.Int64(),
From: sender,
To: recipient,
Details: transferAuditDetails,
Details: details,
}); err != nil {
log.Error("could not send transfer transaction for audit",
zap.String("error", err.Error()),
@ -197,6 +202,10 @@ func (a auditSettlementDeps) Transfer(sender, recipient *owner.ID, amount *big.I
log.Debug("transfer transaction for audit was successfully sent")
}
func (a auditSettlementDeps) Transfer(sender, recipient *owner.ID, amount *big.Int) {
a.transfer(sender, recipient, amount, transferAuditDetails)
}
func (s *auditSettlementCalculator) ProcessAuditSettlements(epoch uint64) {
(*audit.Calculator)(s).Calculate(&audit.CalculatePrm{
Epoch: epoch,