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

storagenode/heldamount: heldhistory reworked to all satellites

Browse Source 
Change-Id: I8d7707fddfbdc52d29951a8a002978c7fbb07049
This commit is contained in:
Qweder93 2020-05-27 19:27:28 +03:00 committed by Nikolai Siedov
parent 4ad163de2f
commit 73214c6d1c
10 changed files with 103 additions and 166 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
storagenode/console/consoleapi/heldamount.go
View File

@ -152,28 +152,15 @@ func (heldAmount *HeldAmount) PayStubPeriod(w http.ResponseWriter, r *http.Reque
}
}
// HeldbackHistory returns heldback for each % period for specific satellite.
func (heldAmount *HeldAmount) HeldbackHistory(w http.ResponseWriter, r *http.Request) {
// HeldHistory returns held amount for each % period for all satellites.
func (heldAmount *HeldAmount) HeldHistory(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
var err error
defer mon.Task()(&ctx)(&err)
w.Header().Set(contentType, applicationJSON)
segmentParams := mux.Vars(r)
id, ok := segmentParams["id"]
if !ok {
heldAmount.serveJSONError(w, http.StatusBadRequest, ErrNotificationsAPI.Wrap(err))
return
}
satelliteID, err := storj.NodeIDFromString(id)
if err != nil {
heldAmount.serveJSONError(w, http.StatusBadRequest, ErrHeldAmountAPI.Wrap(err))
return
}
heldbackHistory, err := heldAmount.service.AllHeldbackHistory(ctx, satelliteID)
heldbackHistory, err := heldAmount.service.AllHeldbackHistory(ctx)
if err != nil {
heldAmount.serveJSONError(w, http.StatusInternalServerError, ErrHeldAmountAPI.Wrap(err))
return

44
storagenode/console/consoleapi/heldamount_test.go
View File

@ -4,6 +4,7 @@
package consoleapi_test
import (
"context"
"encoding/json"
"fmt"
"io/ioutil"
@ -17,6 +18,7 @@ import (
"storj.io/common/testcontext"
"storj.io/storj/private/testplanet"
"storj.io/storj/storagenode/heldamount"
"storj.io/storj/storagenode/reputation"
)
func TestHeldAmountApi(t *testing.T) {
@ -30,6 +32,7 @@ func TestHeldAmountApi(t *testing.T) {
sno := planet.StorageNodes[0]
console := sno.Console
heldAmountDB := sno.DB.HeldAmount()
reputationDB := sno.DB.Reputation()
baseURL := fmt.Sprintf("http://%s/api/heldamount", console.Listener.Addr())
period := "2020-03"
@ -342,20 +345,31 @@ func TestHeldAmountApi(t *testing.T) {
})
t.Run("test HeldbackHistory", func(t *testing.T) {
err = reputationDB.Store(context.Background(), reputation.Stats{
SatelliteID: satellite.ID(),
JoinedAt: time.Now().UTC().AddDate(0, -2, 0),
})
require.NoError(t, err)
// should return all heldback history inserted earlier
url := fmt.Sprintf("%s/heldback/%s", baseURL, satellite.ID().String())
url := fmt.Sprintf("%s/heldhistory", baseURL)
res, err := http.Get(url)
require.NoError(t, err)
require.NotNil(t, res)
require.Equal(t, http.StatusOK, res.StatusCode)
period75 := heldamount.HeldbackPeriod{
PercentageRate: 75,
Held: paystub2.Held + paystub3.Held,
held := heldamount.HeldHistory{
SatelliteID: satellite.ID(),
SatelliteName: satellite.URL(),
Age: 2,
FirstPeriod: 28,
SecondPeriod: 0,
ThirdPeriod: 0,
FourthPeriod: 0,
}
var periods []heldamount.HeldbackPeriod
periods = append(periods, period75)
var periods []heldamount.HeldHistory
periods = append(periods, held)
expected, err := json.Marshal(periods)
require.NoError(t, err)
@ -366,25 +380,7 @@ func TestHeldAmountApi(t *testing.T) {
}()
body, err := ioutil.ReadAll(res.Body)
require.NoError(t, err)
require.Equal(t, string(expected)+"\n", string(body))
// should return 400 because of bad period.
url = fmt.Sprintf("%s/heldback/%s", baseURL, satellite.ID().String()+"11")
res2, err := http.Get(url)
require.NoError(t, err)
require.NotNil(t, res2)
require.Equal(t, http.StatusBadRequest, res2.StatusCode)
defer func() {
err = res2.Body.Close()
require.NoError(t, err)
}()
body2, err := ioutil.ReadAll(res2.Body)
require.NoError(t, err)
require.Equal(t, "{\"error\":\"heldAmount console web error: node ID error: checksum error\"}\n", string(body2))
})
t.Run("test Periods", func(t *testing.T) {

2
storagenode/console/consoleapi/storagenode_test.go
View File

@ -148,6 +148,7 @@ func TestStorageNodeApi(t *testing.T) {
expected, err := json.Marshal(heldamount.EstimatedPayout{
CurrentMonthEstimatedAmount: estimated1 + estimated2,
CurrentMonthHeld: int64(sum1 + sum2 - sum1AfterHeld - sum2AfterHeld),
PreviousMonthPayout: heldamount.PayoutMonthly{
EgressBandwidth: 0,
EgressPayout: 0,
@ -177,6 +178,7 @@ func TestStorageNodeApi(t *testing.T) {
expected2, err := json.Marshal(heldamount.EstimatedPayout{
CurrentMonthEstimatedAmount: estimated1,
CurrentMonthHeld: int64(sum1 - sum1AfterHeld),
PreviousMonthPayout: heldamount.PayoutMonthly{
EgressBandwidth: 0,
EgressPayout: 0,

2
storagenode/console/consoleserver/server.go
View File

@ -82,7 +82,7 @@ func NewServer(logger *zap.Logger, assets http.FileSystem, notifications *notifi
heldAmountRouter.StrictSlash(true)
heldAmountRouter.HandleFunc("/paystubs/{period}", heldAmountController.PayStubMonthly).Methods(http.MethodGet)
heldAmountRouter.HandleFunc("/paystubs/{start}/{end}", heldAmountController.PayStubPeriod).Methods(http.MethodGet)
heldAmountRouter.HandleFunc("/heldback/{id}", heldAmountController.HeldbackHistory).Methods(http.MethodGet)
heldAmountRouter.HandleFunc("/heldhistory", heldAmountController.HeldHistory).Methods(http.MethodGet)
heldAmountRouter.HandleFunc("/periods", heldAmountController.HeldAmountPeriods).Methods(http.MethodGet)
if assets != nil {

20
storagenode/console/service.go
View File

@ -372,7 +372,7 @@ func (s *Service) VerifySatelliteID(ctx context.Context, satelliteID storj.NodeI
func (s *Service) GetSatelliteEstimatedPayout(ctx context.Context, satelliteID storj.NodeID) (payout heldamount.EstimatedPayout, err error) {
defer mon.Task()(&ctx)(&err)
currentMonthPayout, err := s.estimatedPayoutCurrentMonth(ctx, satelliteID)
currentMonthPayout, held, err := s.estimatedPayoutCurrentMonth(ctx, satelliteID)
if err != nil {
return heldamount.EstimatedPayout{}, SNOServiceErr.Wrap(err)
}
@ -383,6 +383,7 @@ func (s *Service) GetSatelliteEstimatedPayout(ctx context.Context, satelliteID s
}
payout.CurrentMonthEstimatedAmount = currentMonthPayout
payout.CurrentMonthHeld = held
payout.PreviousMonthPayout = previousMonthPayout
return payout, nil
@ -394,7 +395,7 @@ func (s *Service) GetAllSatellitesEstimatedPayout(ctx context.Context) (payout h
satelliteIDs := s.trust.GetSatellites(ctx)
for i := 0; i < len(satelliteIDs); i++ {
current, err := s.estimatedPayoutCurrentMonth(ctx, satelliteIDs[i])
current, held, err := s.estimatedPayoutCurrentMonth(ctx, satelliteIDs[i])
if err != nil {
return heldamount.EstimatedPayout{}, SNOServiceErr.Wrap(err)
}
@ -405,6 +406,7 @@ func (s *Service) GetAllSatellitesEstimatedPayout(ctx context.Context) (payout h
}
payout.CurrentMonthEstimatedAmount += current
payout.CurrentMonthHeld += held
payout.PreviousMonthPayout.DiskSpaceAmount += previous.DiskSpaceAmount
payout.PreviousMonthPayout.DiskSpace += previous.DiskSpace
payout.PreviousMonthPayout.EgressBandwidth += previous.EgressBandwidth
@ -417,14 +419,14 @@ func (s *Service) GetAllSatellitesEstimatedPayout(ctx context.Context) (payout h
}
// estimatedPayoutCurrentMonth returns estimated payout for current month from specific satellite with current level of load and previous month.
func (s *Service) estimatedPayoutCurrentMonth(ctx context.Context, satelliteID storj.NodeID) (_ int64, err error) {
func (s *Service) estimatedPayoutCurrentMonth(ctx context.Context, satelliteID storj.NodeID) (payout int64, held int64, err error) {
defer mon.Task()(&ctx)(&err)
var totalSum int64
stats, err := s.reputationDB.Get(ctx, satelliteID)
if err != nil {
return 0, SNOServiceErr.Wrap(err)
return 0, 0, SNOServiceErr.Wrap(err)
}
heldRate := s.getHeldRate(stats.JoinedAt)
@ -434,12 +436,12 @@ func (s *Service) estimatedPayoutCurrentMonth(ctx context.Context, satelliteID s
priceModel, err := s.pricingDB.Get(ctx, satelliteID)
if err != nil {
return 0, SNOServiceErr.Wrap(err)
return 0, 0, SNOServiceErr.Wrap(err)
}
bandwidthDaily, err := s.bandwidthDB.GetDailySatelliteRollups(ctx, satelliteID, from, to)
if err != nil {
return 0, SNOServiceErr.Wrap(err)
return 0, 0, SNOServiceErr.Wrap(err)
}
for i := 0; i < len(bandwidthDaily); i++ {
@ -451,7 +453,7 @@ func (s *Service) estimatedPayoutCurrentMonth(ctx context.Context, satelliteID s
storageDaily, err := s.storageUsageDB.GetDaily(ctx, satelliteID, from, to)
if err != nil {
return 0, SNOServiceErr.Wrap(err)
return 0, 0, SNOServiceErr.Wrap(err)
}
for j := 0; j < len(storageDaily); j++ {
@ -462,7 +464,9 @@ func (s *Service) estimatedPayoutCurrentMonth(ctx context.Context, satelliteID s
day := int64(time.Now().UTC().Day())
amount := totalSum - (totalSum*heldRate)/100
return amount * int64(to.Day()) / day, nil
payout = amount * int64(to.Day()) / day
held = totalSum * heldRate / 100
return payout, held, nil
}
// estimatedPayoutPreviousMonth returns estimated payout data for previous month from specific satellite.

61
storagenode/heldamount/db_test.go
View File

@ -178,7 +178,8 @@ func TestHeldAmountDB(t *testing.T) {
func TestSatellitePayStubPeriodCached(t *testing.T) {
storagenodedbtest.Run(t, func(ctx *testcontext.Context, t *testing.T, db storagenode.DB) {
heldAmountDB := db.HeldAmount()
service := heldamount.NewService(nil, heldAmountDB, rpc.Dialer{}, nil)
reputationDB := db.Reputation()
service := heldamount.NewService(nil, heldAmountDB, reputationDB, rpc.Dialer{}, nil)
payStub := heldamount.PayStub{
SatelliteID: storj.NodeID{1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1},
@ -226,7 +227,8 @@ func TestSatellitePayStubPeriodCached(t *testing.T) {
func TestAllPayStubPeriodCached(t *testing.T) {
storagenodedbtest.Run(t, func(ctx *testcontext.Context, t *testing.T, db storagenode.DB) {
heldAmountDB := db.HeldAmount()
service := heldamount.NewService(nil, heldAmountDB, rpc.Dialer{}, nil)
reputationDB := db.Reputation()
service := heldamount.NewService(nil, heldAmountDB, reputationDB, rpc.Dialer{}, nil)
payStub := heldamount.PayStub{
SatelliteID: storj.NodeID{1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1},
@ -277,58 +279,3 @@ func TestAllPayStubPeriodCached(t *testing.T) {
require.Equal(t, 0, len(payStubs))
})
}
func TestAllHeldbackHistory(t *testing.T) {
storagenodedbtest.Run(t, func(ctx *testcontext.Context, t *testing.T, db storagenode.DB) {
heldAmountDB := db.HeldAmount()
service := heldamount.NewService(nil, heldAmountDB, rpc.Dialer{}, nil)
id := storj.NodeID{1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1}
payStub := heldamount.PayStub{
SatelliteID: id,
Created: time.Now().UTC(),
Codes: "code",
UsageAtRest: 1,
UsageGet: 2,
UsagePut: 3,
UsageGetRepair: 4,
UsagePutRepair: 5,
UsageGetAudit: 6,
CompAtRest: 7,
CompGet: 8,
CompPut: 9,
CompGetRepair: 10,
CompPutRepair: 11,
CompGetAudit: 12,
SurgePercent: 13,
Held: 14,
Owed: 15,
Disposed: 16,
Paid: 17,
}
for j := 1; j < 5; j++ {
payStub.Period = fmt.Sprintf("2020-0%d", j)
err := heldAmountDB.StorePayStub(ctx, payStub)
require.NoError(t, err)
}
heldback, err := service.AllHeldbackHistory(ctx, id)
require.NoError(t, err)
var expectedResult []heldamount.HeldbackPeriod
period75 := heldamount.HeldbackPeriod{
PercentageRate: 75,
Held: payStub.Held * 3,
}
period50 := heldamount.HeldbackPeriod{
PercentageRate: 50,
Held: payStub.Held,
}
expectedResult = append(expectedResult, period75, period50)
require.Equal(t, expectedResult, heldback)
})
}

7
storagenode/heldamount/heldamount.go
View File

@ -23,7 +23,7 @@ type DB interface {
// AllPayStubs retrieves paystub data from all satellites in specific period from DB.
AllPayStubs(ctx context.Context, period string) ([]PayStub, error)
// SatellitesHeldbackHistory retrieves heldback history for specific satellite from DB.
SatellitesHeldbackHistory(ctx context.Context, satelliteID storj.NodeID) ([]Heldback, error)
SatellitesHeldbackHistory(ctx context.Context, satelliteID storj.NodeID) ([]AmountPeriod, error)
// SatellitePeriods retrieves all periods for concrete satellite in which we have some heldamount data.
SatellitePeriods(ctx context.Context, satelliteID storj.NodeID) ([]string, error)
// AllPeriods retrieves all periods in which we have some heldamount data.
@ -58,8 +58,8 @@ type PayStub struct {
Paid int64 `json:"paid"`
}
// Heldback is node's heldback amount for period.
type Heldback struct {
// AmountPeriod is node's held amount for period.
type AmountPeriod struct {
Period string `json:"period"`
Held int64 `json:"held"`
}
@ -67,6 +67,7 @@ type Heldback struct {
// EstimatedPayout contains amount in cents of estimated payout for current and previous months.
type EstimatedPayout struct {
CurrentMonthEstimatedAmount int64 `json:"currentAmount"`
CurrentMonthHeld int64 `json:"currentHeld"`
PreviousMonthPayout PayoutMonthly `json:"previousPayout"`
}

107
storagenode/heldamount/service.go
View File

@ -18,6 +18,7 @@ import (
"storj.io/common/rpc/rpcstatus"
"storj.io/storj/pkg/storj"
"storj.io/storj/private/date"
"storj.io/storj/storagenode/reputation"
"storj.io/storj/storagenode/trust"
)
@ -52,19 +53,21 @@ func (c *Client) Close() error {
type Service struct {
log *zap.Logger
db DB
db DB
reputationDB reputation.DB
dialer rpc.Dialer
trust *trust.Pool
}
// NewService creates new instance of service
func NewService(log *zap.Logger, db DB, dialer rpc.Dialer, trust *trust.Pool) *Service {
func NewService(log *zap.Logger, db DB, reputationDB reputation.DB, dialer rpc.Dialer, trust *trust.Pool) *Service {
return &Service{
log: log,
db: db,
dialer: dialer,
trust: trust,
log: log,
db: db,
reputationDB: reputationDB,
dialer: dialer,
trust: trust,
}
}
@ -253,62 +256,58 @@ func (service *Service) AllPeriods(ctx context.Context) (_ []string, err error)
return service.db.AllPeriods(ctx)
}
// HeldbackPeriod amount of held for specific percent rate period.
type HeldbackPeriod struct {
PercentageRate int
Held int64
// HeldHistory amount of held for specific percent rate period.
type HeldHistory struct {
SatelliteID storj.NodeID `json:"satelliteID"`
SatelliteName string `json:"satelliteName"`
Age int64 `json:"age"`
FirstPeriod int64 `json:"firstPeriod"`
SecondPeriod int64 `json:"secondPeriod"`
ThirdPeriod int64 `json:"thirdPeriod"`
FourthPeriod int64 `json:"fourthPeriod"`
}
// AllHeldbackHistory retrieves heldback history for all specific satellite from storagenode database.
func (service *Service) AllHeldbackHistory(ctx context.Context, id storj.NodeID) (result []HeldbackPeriod, err error) {
defer mon.Task()(&ctx, &id)(&err)
// AllHeldbackHistory retrieves heldback history for all satellites from storagenode database.
func (service *Service) AllHeldbackHistory(ctx context.Context) (result []HeldHistory, err error) {
defer mon.Task()(&ctx)(&err)
heldback, err := service.db.SatellitesHeldbackHistory(ctx, id)
if err != nil {
return nil, ErrHeldAmountService.Wrap(err)
}
satellites := service.trust.GetSatellites(ctx)
for i := 0; i < len(satellites); i++ {
var history HeldHistory
var total75, total50, total25, total0 int64
for i, t := range heldback {
switch i {
case 0, 1, 2:
total75 += t.Held
case 3, 4, 5:
total50 += t.Held
case 6, 7, 8:
total25 += t.Held
default:
total0 += t.Held
heldback, err := service.db.SatellitesHeldbackHistory(ctx, satellites[i])
if err != nil {
return nil, ErrHeldAmountService.Wrap(err)
}
}
period75percent := HeldbackPeriod{
PercentageRate: 75,
Held: total75,
}
period50percent := HeldbackPeriod{
PercentageRate: 50,
Held: total50,
}
period25percent := HeldbackPeriod{
PercentageRate: 25,
Held: total25,
}
period0percent := HeldbackPeriod{
PercentageRate: 0,
Held: total0,
}
for i, t := range heldback {
switch i {
case 0, 1, 2:
history.FirstPeriod += t.Held
case 3, 4, 5:
history.SecondPeriod += t.Held
case 6, 7, 8:
history.ThirdPeriod += t.Held
default:
history.FourthPeriod += t.Held
}
}
result = append(result, period75percent)
history.SatelliteID = satellites[i]
url, err := service.trust.GetNodeURL(ctx, satellites[i])
if err != nil {
return nil, ErrHeldAmountService.Wrap(err)
}
switch {
case len(heldback) > 3:
result = append(result, period50percent)
case len(heldback) > 6:
result = append(result, period25percent)
case len(heldback) > 9:
result = append(result, period0percent)
stats, err := service.reputationDB.Get(ctx, satellites[i])
if err != nil {
return nil, ErrHeldAmountService.Wrap(err)
}
history.Age = int64(date.MonthsCountSince(stats.JoinedAt))
history.SatelliteName = url.String()
result = append(result, history)
}
return result, nil

1
storagenode/peer.go
View File

@ -526,6 +526,7 @@ func New(log *zap.Logger, full *identity.FullIdentity, db DB, revocationDB exten
peer.Heldamount.Service = heldamount.NewService(
peer.Log.Named("heldamount:service"),
peer.DB.HeldAmount(),
peer.DB.Reputation(),
peer.Dialer,
peer.Storage2.Trust,
)

6
storagenode/storagenodedb/heldamount.go
View File

@ -220,7 +220,7 @@ func (db *heldamountDB) AllPayStubs(ctx context.Context, period string) (_ []hel
}
// SatellitesHeldbackHistory retrieves heldback history for specific satellite.
func (db *heldamountDB) SatellitesHeldbackHistory(ctx context.Context, id storj.NodeID) (_ []heldamount.Heldback, err error) {
func (db *heldamountDB) SatellitesHeldbackHistory(ctx context.Context, id storj.NodeID) (_ []heldamount.AmountPeriod, err error) {
defer mon.Task()(&ctx)(&err)
query := `SELECT
@ -235,9 +235,9 @@ func (db *heldamountDB) SatellitesHeldbackHistory(ctx context.Context, id storj.
defer func() { err = errs.Combine(err, rows.Close()) }()
var heldback []heldamount.Heldback
var heldback []heldamount.AmountPeriod
for rows.Next() {
var held heldamount.Heldback
var held heldamount.AmountPeriod
err := rows.Scan(&held.Period, &held.Held)
if err != nil {