JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

storagenode/payouts: historical payouts use satellitesDB instead of trustPool

Browse Source 
Change-Id: I39f4215f4ebf91bd1b38fbcb5c58e6ba53ceff1b
This commit is contained in:
Qweder93 2021-07-09 15:29:34 +03:00
parent 18c3252025
commit 4a98dd40e2
8 changed files with 61 additions and 151 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
storagenode/multinode/payout_test.go
View File

@ -66,7 +66,7 @@ func TestPayoutsEndpointSummary(t *testing.T) {
require.NoError(t, err)
require.NoError(t, trustPool.Refresh(ctx))
payoutsService, err := payouts.NewService(log, db.Payout(), db.Reputation(), db.Satellites(), nil)
payoutsService, err := payouts.NewService(log, db.Payout(), db.Reputation(), db.Satellites())
require.NoError(t, err)
estimatedPayoutsService := estimatedpayouts.NewService(db.Bandwidth(), db.Reputation(), db.StorageUsage(), db.Pricing(), db.Satellites(), trustPool)
endpoint := multinode.NewPayoutEndpoint(log, service, db.Payout(), estimatedPayoutsService, payoutsService)
@ -159,7 +159,7 @@ func TestPayoutsEndpointEstimations(t *testing.T) {
require.NoError(t, err)
require.NoError(t, trustPool.Refresh(ctx))
payoutsService, err := payouts.NewService(log, db.Payout(), db.Reputation(), db.Satellites(), nil)
payoutsService, err := payouts.NewService(log, db.Payout(), db.Reputation(), db.Satellites())
require.NoError(t, err)
estimatedPayoutsService := estimatedpayouts.NewService(db.Bandwidth(), db.Reputation(), db.StorageUsage(), db.Pricing(), db.Satellites(), trustPool)
endpoint := multinode.NewPayoutEndpoint(log, service, db.Payout(), estimatedPayoutsService, payoutsService)
@ -230,7 +230,7 @@ func TestPayoutsUndistributedEndpoint(t *testing.T) {
require.NoError(t, err)
require.NoError(t, trustPool.Refresh(ctx))
payoutsService, err := payouts.NewService(log, db.Payout(), db.Reputation(), db.Satellites(), nil)
payoutsService, err := payouts.NewService(log, db.Payout(), db.Reputation(), db.Satellites())
require.NoError(t, err)
estimatedPayoutsService := estimatedpayouts.NewService(db.Bandwidth(), db.Reputation(), db.StorageUsage(), db.Pricing(), db.Satellites(), trustPool)
endpoint := multinode.NewPayoutEndpoint(log, service, db.Payout(), estimatedPayoutsService, payoutsService)

7
storagenode/payouts/db_test.go
View File

@ -10,6 +10,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/zap/zaptest"
"storj.io/common/testcontext"
"storj.io/common/testrand"
@ -211,7 +212,8 @@ func TestSatellitePayStubPeriodCached(t *testing.T) {
heldAmountDB := db.Payout()
reputationDB := db.Reputation()
satellitesDB := db.Satellites()
service, err := payouts.NewService(nil, heldAmountDB, reputationDB, satellitesDB, nil)
log := zaptest.NewLogger(t)
service, err := payouts.NewService(log, heldAmountDB, reputationDB, satellitesDB)
require.NoError(t, err)
payStub := payouts.PayStub{
@ -262,7 +264,8 @@ func TestAllPayStubPeriodCached(t *testing.T) {
heldAmountDB := db.Payout()
reputationDB := db.Reputation()
satellitesDB := db.Satellites()
service, err := payouts.NewService(nil, heldAmountDB, reputationDB, satellitesDB, nil)
log := zaptest.NewLogger(t)
service, err := payouts.NewService(log, heldAmountDB, reputationDB, satellitesDB)
require.NoError(t, err)
payStub := payouts.PayStub{

94
storagenode/payouts/service.go
View File

@ -20,7 +20,6 @@ import (
"storj.io/storj/private/date"
"storj.io/storj/storagenode/reputation"
"storj.io/storj/storagenode/satellites"
"storj.io/storj/storagenode/trust"
)
var (
@ -39,28 +38,18 @@ var (
type Service struct {
log *zap.Logger
stefanSatellite storj.NodeID
db DB
reputationDB reputation.DB
satellitesDB satellites.DB
trust *trust.Pool
}
// NewService creates new instance of service.
func NewService(log *zap.Logger, db DB, reputationDB reputation.DB, satelliteDB satellites.DB, trust *trust.Pool) (_ *Service, err error) {
id, err := storj.NodeIDFromString("118UWpMCHzs6CvSgWd9BfFVjw5K9pZbJjkfZJexMtSkmKxvvAW")
if err != nil {
return &Service{}, err
}
func NewService(log *zap.Logger, db DB, reputationDB reputation.DB, satelliteDB satellites.DB) (_ *Service, err error) {
return &Service{
log: log,
stefanSatellite: id,
db: db,
reputationDB: reputationDB,
satellitesDB: satelliteDB,
trust: trust,
log: log,
db: db,
reputationDB: reputationDB,
satellitesDB: satelliteDB,
}, nil
}
@ -173,13 +162,14 @@ func (service *Service) AllPeriods(ctx context.Context) (_ []string, err error)
// AllHeldbackHistory retrieves heldback history for all satellites from storagenode database.
func (service *Service) AllHeldbackHistory(ctx context.Context) (result []SatelliteHeldHistory, err error) {
defer mon.Task()(&ctx)(&err)
satellitesIDs := service.trust.GetSatellites(ctx)
satellitesIDs = append(satellitesIDs, service.stefanSatellite)
for i := 0; i < len(satellitesIDs); i++ {
satellitesURLs, err := service.satellitesDB.GetSatellitesURLs(ctx)
if err != nil {
return nil, ErrPayoutService.Wrap(err)
}
for _, satelliteURL := range satellitesURLs {
var history SatelliteHeldHistory
helds, err := service.db.SatellitesHeldbackHistory(ctx, satellitesIDs[i])
helds, err := service.db.SatellitesHeldbackHistory(ctx, satelliteURL.ID)
if err != nil {
return nil, ErrPayoutService.Wrap(err)
}
@ -188,7 +178,7 @@ func (service *Service) AllHeldbackHistory(ctx context.Context) (result []Satell
continue
}
disposed, err := service.db.SatellitesDisposedHistory(ctx, satellitesIDs[i])
disposed, err := service.db.SatellitesDisposedHistory(ctx, satelliteURL.ID)
if err != nil {
return nil, ErrPayoutService.Wrap(err)
}
@ -209,18 +199,14 @@ func (service *Service) AllHeldbackHistory(ctx context.Context) (result []Satell
}
history.TotalDisposed = disposed
history.SatelliteID = satellitesIDs[i]
history.SatelliteName = "stefan-benten"
history.SatelliteID = satelliteURL.ID
history.SatelliteName = satelliteURL.Address
if satellitesIDs[i] != service.stefanSatellite {
url, err := service.trust.GetNodeURL(ctx, satellitesIDs[i])
if err != nil {
return nil, ErrPayoutService.Wrap(err)
}
history.SatelliteName = url.Address
if satelliteURL.Address == "" {
history.SatelliteName = satelliteURL.ID.String()
}
stats, err := service.reputationDB.Get(ctx, satellitesIDs[i])
stats, err := service.reputationDB.Get(ctx, satelliteURL.ID)
if err != nil {
return nil, ErrPayoutService.Wrap(err)
}
@ -236,12 +222,14 @@ func (service *Service) AllHeldbackHistory(ctx context.Context) (result []Satell
func (service *Service) AllSatellitesPayoutPeriod(ctx context.Context, period string) (result []SatellitePayoutForPeriod, err error) {
defer mon.Task()(&ctx)(&err)
satelliteIDs := service.trust.GetSatellites(ctx)
satelliteIDs = append(satelliteIDs, service.stefanSatellite)
for i := 0; i < len(satelliteIDs); i++ {
satelliteURLs, err := service.satellitesDB.GetSatellitesURLs(ctx)
if err != nil {
return nil, ErrPayoutService.Wrap(err)
}
for _, satelliteURL := range satelliteURLs {
var payoutForPeriod SatellitePayoutForPeriod
paystub, err := service.db.GetPayStub(ctx, satelliteIDs[i], period)
paystub, err := service.db.GetPayStub(ctx, satelliteURL.ID, period)
if err != nil {
if ErrNoPayStubForPeriod.Has(err) {
continue
@ -249,19 +237,19 @@ func (service *Service) AllSatellitesPayoutPeriod(ctx context.Context, period st
return nil, ErrPayoutService.Wrap(err)
}
receipt, err := service.db.GetReceipt(ctx, satelliteIDs[i], period)
receipt, err := service.db.GetReceipt(ctx, satelliteURL.ID, period)
if err != nil {
if !ErrNoPayStubForPeriod.Has(err) {
return nil, ErrPayoutService.Wrap(err)
}
}
stats, err := service.reputationDB.Get(ctx, satelliteIDs[i])
stats, err := service.reputationDB.Get(ctx, satelliteURL.ID)
if err != nil {
return nil, ErrPayoutService.Wrap(err)
}
satellite, err := service.satellitesDB.GetSatellite(ctx, satelliteIDs[i])
satellite, err := service.satellitesDB.GetSatellite(ctx, satelliteURL.ID)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
payoutForPeriod.IsExitComplete = false
@ -270,15 +258,6 @@ func (service *Service) AllSatellitesPayoutPeriod(ctx context.Context, period st
return nil, ErrPayoutService.Wrap(err)
}
if satelliteIDs[i] != service.stefanSatellite {
url, err := service.trust.GetNodeURL(ctx, satelliteIDs[i])
if err != nil {
return nil, ErrPayoutService.Wrap(err)
}
payoutForPeriod.SatelliteURL = url.Address
}
if satellite.Status == satellites.ExitSucceeded {
payoutForPeriod.IsExitComplete = true
}
@ -304,12 +283,17 @@ func (service *Service) AllSatellitesPayoutPeriod(ctx context.Context, period st
payoutForPeriod.Age = int64(date.MonthsCountSince(stats.JoinedAt))
payoutForPeriod.Disposed = paystub.Disposed
payoutForPeriod.Earned = earned
payoutForPeriod.SatelliteID = satelliteIDs[i].String()
payoutForPeriod.SatelliteID = satelliteURL.ID.String()
payoutForPeriod.SatelliteURL = satelliteURL.Address
payoutForPeriod.SurgePercent = paystub.SurgePercent
payoutForPeriod.Paid = paystub.Paid
payoutForPeriod.HeldPercent = heldPercent
payoutForPeriod.Distributed = paystub.Distributed
if satelliteURL.Address == "" {
payoutForPeriod.SatelliteURL = satelliteURL.ID.String()
}
result = append(result, payoutForPeriod)
}
@ -325,20 +309,22 @@ func (service *Service) HeldAmountHistory(ctx context.Context) (_ []HeldAmountHi
return nil, ErrPayoutService.Wrap(err)
}
trustedSatellites := service.trust.GetSatellites(ctx)
for _, trustedSatellite := range trustedSatellites {
satelliteURLs, err := service.satellitesDB.GetSatellitesURLs(ctx)
if err != nil {
return nil, ErrPayoutService.Wrap(err)
}
for _, satelliteURL := range satelliteURLs {
var found bool
for _, satelliteHeldHistory := range heldHistory {
if trustedSatellite.Compare(satelliteHeldHistory.SatelliteID) == 0 {
if satelliteURL.ID.Compare(satelliteHeldHistory.SatelliteID) == 0 {
found = true
break
}
}
if !found {
heldHistory = append(heldHistory, HeldAmountHistory{
SatelliteID: trustedSatellite,
SatelliteID: satelliteURL.ID,
})
}
}

92
storagenode/payouts/service_test.go
View File

@ -4,22 +4,16 @@
package payouts_test
import (
"context"
"errors"
"sync"
"testing"
"github.com/stretchr/testify/require"
"go.uber.org/zap/zaptest"
"storj.io/common/identity"
"storj.io/common/storj"
"storj.io/common/testcontext"
"storj.io/common/testrand"
"storj.io/storj/storagenode"
"storj.io/storj/storagenode/payouts"
"storj.io/storj/storagenode/storagenodedb/storagenodedbtest"
"storj.io/storj/storagenode/trust"
)
func TestServiceHeldAmountHistory(t *testing.T) {
@ -27,42 +21,17 @@ func TestServiceHeldAmountHistory(t *testing.T) {
log := zaptest.NewLogger(t)
payoutsDB := db.Payout()
satellitesDB := db.Satellites()
source := &fakeSource{}
pool, err := trust.NewPool(log, newFakeIdentityResolver(), trust.Config{
Sources: []trust.Source{source},
CachePath: ctx.File("trust-cache.json"),
}, satellitesDB)
require.NoError(t, err)
satelliteID1 := testrand.NodeID()
satelliteID2 := testrand.NodeID()
satelliteID3 := testrand.NodeID()
// populate pool
source.entries = []trust.Entry{
{
SatelliteURL: trust.SatelliteURL{
ID: satelliteID1,
Host: "foo.test",
Port: 7777,
},
},
{
SatelliteURL: trust.SatelliteURL{
ID: satelliteID2,
Host: "bar.test",
Port: 7777,
},
},
{
SatelliteURL: trust.SatelliteURL{
ID: satelliteID3,
Host: "baz.test",
Port: 7777,
},
},
}
require.NoError(t, pool.Refresh(context.Background()))
err := satellitesDB.SetAddress(ctx, satelliteID1, "foo.test:7777")
require.NoError(t, err)
err = satellitesDB.SetAddress(ctx, satelliteID2, "bar.test:7777")
require.NoError(t, err)
err = satellitesDB.SetAddress(ctx, satelliteID3, "baz.test:7777")
require.NoError(t, err)
// add paystubs
paystubs := []payouts.PayStub{
@ -124,7 +93,7 @@ func TestServiceHeldAmountHistory(t *testing.T) {
},
}
service, err := payouts.NewService(log, payoutsDB, db.Reputation(), db.Satellites(), pool)
service, err := payouts.NewService(log, payoutsDB, db.Reputation(), db.Satellites())
require.NoError(t, err)
history, err := service.HeldAmountHistory(ctx)
@ -132,50 +101,3 @@ func TestServiceHeldAmountHistory(t *testing.T) {
require.ElementsMatch(t, expected, history)
})
}
type fakeSource struct {
name string
static bool
entries []trust.Entry
err error
}
func (s *fakeSource) String() string {
return s.name
}
func (s *fakeSource) Static() bool {
return s.static
}
func (s *fakeSource) FetchEntries(context.Context) ([]trust.Entry, error) {
return s.entries, s.err
}
type fakeIdentityResolver struct {
mu sync.Mutex
identities map[storj.NodeURL]*identity.PeerIdentity
}
func newFakeIdentityResolver() *fakeIdentityResolver {
return &fakeIdentityResolver{
identities: make(map[storj.NodeURL]*identity.PeerIdentity),
}
}
func (resolver *fakeIdentityResolver) SetIdentity(url storj.NodeURL, identity *identity.PeerIdentity) {
resolver.mu.Lock()
defer resolver.mu.Unlock()
resolver.identities[url] = identity
}
func (resolver *fakeIdentityResolver) ResolveIdentity(ctx context.Context, url storj.NodeURL) (*identity.PeerIdentity, error) {
resolver.mu.Lock()
defer resolver.mu.Unlock()
identity := resolver.identities[url]
if identity == nil {
return nil, errors.New("no identity")
}
return identity, nil
}

1
storagenode/peer.go
View File

@ -569,7 +569,6 @@ func New(log *zap.Logger, full *identity.FullIdentity, db DB, revocationDB exten
peer.DB.Payout(),
peer.DB.Reputation(),
peer.DB.Satellites(),
peer.Storage2.Trust,
)
if err != nil {
return nil, errs.Combine(err, peer.Close())

4
storagenode/satellites/db_test.go
View File

@ -22,14 +22,14 @@ func TestSatellitesDB(t *testing.T) {
err := satellitesDB.SetAddress(ctx, id, "test_addr1")
require.NoError(t, err)
satellites, err := satellitesDB.GetSatellitesUrls(ctx)
satellites, err := satellitesDB.GetSatellitesURLs(ctx)
require.NoError(t, err)
require.Equal(t, satellites[0].Address, "test_addr1")
err = satellitesDB.SetAddress(ctx, id, "test_addr2")
require.NoError(t, err)
satellites, err = satellitesDB.GetSatellitesUrls(ctx)
satellites, err = satellitesDB.GetSatellitesURLs(ctx)
require.NoError(t, err)
require.Equal(t, satellites[0].Address, "test_addr2")
})

4
storagenode/satellites/satellites.go
View File

@ -52,8 +52,8 @@ type DB interface {
SetAddress(ctx context.Context, satelliteID storj.NodeID, address string) error
// GetSatellite retrieves that satellite by ID
GetSatellite(ctx context.Context, satelliteID storj.NodeID) (satellite Satellite, err error)
// GetSatellitesUrls retrieves all satellite's id and urls.
GetSatellitesUrls(ctx context.Context) (satelliteURLs []storj.NodeURL, err error)
// GetSatellitesURLs retrieves all satellite's id and urls.
GetSatellitesURLs(ctx context.Context) (satelliteURLs []storj.NodeURL, err error)
// InitiateGracefulExit updates the database to reflect the beginning of a graceful exit
InitiateGracefulExit(ctx context.Context, satelliteID storj.NodeID, intitiatedAt time.Time, startingDiskUsage int64) error
// CancelGracefulExit removes that satellite by ID

4
storagenode/storagenodedb/satellites.go
View File

@ -59,8 +59,8 @@ func (db *satellitesDB) GetSatellite(ctx context.Context, satelliteID storj.Node
return satellite, rows.Err()
}
// GetSatellitesUrls retrieves all satellite's id and urls.
func (db *satellitesDB) GetSatellitesUrls(ctx context.Context) (satelliteURLs []storj.NodeURL, err error) {
// GetSatellitesURLs retrieves all satellite's id and urls.
func (db *satellitesDB) GetSatellitesURLs(ctx context.Context) (satelliteURLs []storj.NodeURL, err error) {
defer mon.Task()(&ctx)(&err)
query := `SELECT