cmd/satellite: add graceful exit reports command to satellite CLI (#3300)
* update lock file and add comment * add created at and bytes transferred * cleanup * rename db func to GetGracefulExitNodesByTimeFrame * fix flag * split into two overlay functions * := to = * fix test * add node not found error class * fix overlay test * suggested test changes * review suggestions * get exit status from overlay.Get() * check rows.Err * fix panic when ExitFinishedAt is nil * fix comments in cmdGracefulExit
This commit is contained in:
Maximillian von Briesen
GitHub
parent
51d5d8656a
commit
abb567f6ae
100
cmd/satellite/gracefulexit.go
Normal file
100
cmd/satellite/gracefulexit.go
Normal file
@ -0,0 +1,100 @@
|
||||
// Copyright (C) 2019 Storj Labs, Inc.
|
||||
// See LICENSE for copying information.
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/csv"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/zeebo/errs"
|
||||
"go.uber.org/zap"
|
||||
|
||||
"storj.io/storj/pkg/storj"
|
||||
"storj.io/storj/satellite/gracefulexit"
|
||||
"storj.io/storj/satellite/satellitedb"
|
||||
)
|
||||
|
||||
// generateGracefulExitCSV creates a report with graceful exit data for exiting or exited nodes in a given period
|
||||
func generateGracefulExitCSV(ctx context.Context, completed bool, start time.Time, end time.Time, output io.Writer) error {
|
||||
db, err := satellitedb.New(zap.L().Named("db"), gracefulExitCfg.Database)
|
||||
if err != nil {
|
||||
return errs.New("error connecting to master database on satellite: %+v", err)
|
||||
}
|
||||
defer func() {
|
||||
err = errs.Combine(err, db.Close())
|
||||
}()
|
||||
|
||||
var nodeIDs storj.NodeIDList
|
||||
if completed {
|
||||
nodeIDs, err = db.OverlayCache().GetGracefulExitCompletedByTimeFrame(ctx, start, end)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
nodeIDs, err = db.OverlayCache().GetGracefulExitIncompleteByTimeFrame(ctx, start, end)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
w := csv.NewWriter(output)
|
||||
headers := []string{
|
||||
"nodeID",
|
||||
"walletAddress",
|
||||
"nodeCreationDate",
|
||||
"initiatedGracefulExit",
|
||||
"completedGracefulExit",
|
||||
"transferredGB",
|
||||
}
|
||||
if err := w.Write(headers); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, id := range nodeIDs {
|
||||
node, err := db.OverlayCache().Get(ctx, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
exitProgress, err := db.GracefulExit().GetProgress(ctx, id)
|
||||
if gracefulexit.ErrNodeNotFound.Has(err) {
|
||||
exitProgress = &gracefulexit.Progress{}
|
||||
} else if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
exitStatus := node.ExitStatus
|
||||
exitFinished := ""
|
||||
if exitStatus.ExitFinishedAt != nil {
|
||||
exitFinished = exitStatus.ExitFinishedAt.Format("2006-01-02")
|
||||
}
|
||||
nextRow := []string{
|
||||
node.Id.String(),
|
||||
node.Operator.Wallet,
|
||||
node.CreatedAt.Format("2006-01-02"),
|
||||
exitStatus.ExitInitiatedAt.Format("2006-01-02"),
|
||||
exitFinished,
|
||||
strconv.FormatInt(exitProgress.BytesTransferred, 10),
|
||||
}
|
||||
if err := w.Write(nextRow); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := w.Error(); err != nil {
|
||||
return err
|
||||
}
|
||||
w.Flush()
|
||||
if output != os.Stdout {
|
||||
if completed {
|
||||
fmt.Println("Generated report for nodes that have completed graceful exit.")
|
||||
} else {
|
||||
fmt.Println("Generated report for nodes that are in the process of graceful exiting.")
|
||||
}
|
||||
}
|
||||
return err
|
||||
}
|
||||
@ -79,6 +79,13 @@ var (
|
||||
Args: cobra.MinimumNArgs(3),
|
||||
RunE: cmdValueAttribution,
|
||||
}
|
||||
gracefulExitCmd = &cobra.Command{
|
||||
Use: "graceful-exit [start] [end]",
|
||||
Short: "Generate a graceful exit report",
|
||||
Long: "Generate a node usage report for a given period to use for payments. Format dates using YYYY-MM-DD",
|
||||
Args: cobra.MinimumNArgs(2),
|
||||
RunE: cmdGracefulExit,
|
||||
}
|
||||
|
||||
runCfg Satellite
|
||||
setupCfg Satellite
|
||||
@ -95,6 +102,11 @@ var (
|
||||
Database string `help:"satellite database connection string" releaseDefault:"postgres://" devDefault:"postgres://"`
|
||||
Output string `help:"destination of report output" default:""`
|
||||
}
|
||||
gracefulExitCfg struct {
|
||||
Database string `help:"satellite database connection string" releaseDefault:"postgres://" devDefault:"sqlite3://$CONFDIR/master.db"`
|
||||
Output string `help:"destination of report output" default:""`
|
||||
Completed bool `help:"whether to output (initiated and completed) or (initiated and not completed)" default:"false"`
|
||||
}
|
||||
confDir string
|
||||
identityDir string
|
||||
)
|
||||
@ -112,11 +124,13 @@ func init() {
|
||||
rootCmd.AddCommand(reportsCmd)
|
||||
reportsCmd.AddCommand(nodeUsageCmd)
|
||||
reportsCmd.AddCommand(partnerAttributionCmd)
|
||||
reportsCmd.AddCommand(gracefulExitCmd)
|
||||
process.Bind(runCmd, &runCfg, defaults, cfgstruct.ConfDir(confDir), cfgstruct.IdentityDir(identityDir))
|
||||
process.Bind(runAPICmd, &runCfg, defaults, cfgstruct.ConfDir(confDir), cfgstruct.IdentityDir(identityDir))
|
||||
process.Bind(setupCmd, &setupCfg, defaults, cfgstruct.ConfDir(confDir), cfgstruct.IdentityDir(identityDir), cfgstruct.SetupMode())
|
||||
process.Bind(qdiagCmd, &qdiagCfg, defaults, cfgstruct.ConfDir(confDir), cfgstruct.IdentityDir(identityDir))
|
||||
process.Bind(nodeUsageCmd, &nodeUsageCfg, defaults, cfgstruct.ConfDir(confDir), cfgstruct.IdentityDir(identityDir))
|
||||
process.Bind(gracefulExitCmd, &gracefulExitCfg, defaults, cfgstruct.ConfDir(confDir), cfgstruct.IdentityDir(identityDir))
|
||||
process.Bind(partnerAttributionCmd, &partnerAttribtionCfg, defaults, cfgstruct.ConfDir(confDir), cfgstruct.IdentityDir(identityDir))
|
||||
}
|
||||
|
||||
@ -301,6 +315,45 @@ func cmdQDiag(cmd *cobra.Command, args []string) (err error) {
|
||||
return w.Flush()
|
||||
}
|
||||
|
||||
func cmdGracefulExit(cmd *cobra.Command, args []string) (err error) {
|
||||
ctx, _ := process.Ctx(cmd)
|
||||
|
||||
layout := "2006-01-02"
|
||||
start, err := time.Parse(layout, args[0])
|
||||
if err != nil {
|
||||
return errs.New("Invalid date format. Please use YYYY-MM-DD")
|
||||
}
|
||||
end, err := time.Parse(layout, args[1])
|
||||
if err != nil {
|
||||
return errs.New("Invalid date format. Please use YYYY-MM-DD")
|
||||
}
|
||||
|
||||
// adding one day to properly account for the entire end day
|
||||
end = end.AddDate(0, 0, 1)
|
||||
|
||||
// ensure that start date is not after end date
|
||||
if start.After(end) {
|
||||
return errs.New("Invalid time period (%v) - (%v)", start, end)
|
||||
}
|
||||
|
||||
// send output to stdout
|
||||
if gracefulExitCfg.Output == "" {
|
||||
return generateGracefulExitCSV(ctx, gracefulExitCfg.Completed, start, end, os.Stdout)
|
||||
}
|
||||
|
||||
// send output to file
|
||||
file, err := os.Create(gracefulExitCfg.Output)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err = errs.Combine(err, file.Close())
|
||||
}()
|
||||
|
||||
return generateGracefulExitCSV(ctx, gracefulExitCfg.Completed, start, end, file)
|
||||
}
|
||||
|
||||
func cmdNodeUsage(cmd *cobra.Command, args []string) (err error) {
|
||||
ctx, _ := process.Ctx(cmd)
|
||||
|
||||
@ -324,7 +377,7 @@ func cmdNodeUsage(cmd *cobra.Command, args []string) (err error) {
|
||||
|
||||
// send output to stdout
|
||||
if nodeUsageCfg.Output == "" {
|
||||
return generateCSV(ctx, start, end, os.Stdout)
|
||||
return generateNodeUsageCSV(ctx, start, end, os.Stdout)
|
||||
}
|
||||
|
||||
// send output to file
|
||||
@ -337,7 +390,7 @@ func cmdNodeUsage(cmd *cobra.Command, args []string) (err error) {
|
||||
err = errs.Combine(err, file.Close())
|
||||
}()
|
||||
|
||||
return generateCSV(ctx, start, end, file)
|
||||
return generateNodeUsageCSV(ctx, start, end, file)
|
||||
}
|
||||
|
||||
func cmdValueAttribution(cmd *cobra.Command, args []string) (err error) {
|
||||
|
||||
@ -19,8 +19,8 @@ import (
|
||||
"storj.io/storj/satellite/satellitedb"
|
||||
)
|
||||
|
||||
// generateCSV creates a report with node usage data for all nodes in a given period which can be used for payments
|
||||
func generateCSV(ctx context.Context, start time.Time, end time.Time, output io.Writer) error {
|
||||
// generateNodeUsageCSV creates a report with node usage data for all nodes in a given period which can be used for payments
|
||||
func generateNodeUsageCSV(ctx context.Context, start time.Time, end time.Time, output io.Writer) error {
|
||||
db, err := satellitedb.New(zap.L().Named("db"), nodeUsageCfg.Database)
|
||||
if err != nil {
|
||||
return errs.New("error connecting to master database on satellite: %+v", err)
|
||||
|
||||
@ -14,6 +14,9 @@ var (
|
||||
// Error is the default error class for graceful exit package.
|
||||
Error = errs.Class("gracefulexit")
|
||||
|
||||
// ErrNodeNotFound is returned if a graceful exit entry for a node does not exist in database
|
||||
ErrNodeNotFound = errs.Class("graceful exit node not found")
|
||||
|
||||
mon = monkit.Package()
|
||||
)
|
||||
|
||||
|
||||
@ -82,3 +82,107 @@ func TestGetExitingNodes(t *testing.T) {
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestGetGracefulExitNodesByTimeframe(t *testing.T) {
|
||||
satellitedbtest.Run(t, func(t *testing.T, db satellite.DB) {
|
||||
ctx := testcontext.New(t)
|
||||
defer ctx.Cleanup()
|
||||
|
||||
cache := db.OverlayCache()
|
||||
exitingToday := make(map[storj.NodeID]bool)
|
||||
exitingLastWeek := make(map[storj.NodeID]bool)
|
||||
exitedToday := make(map[storj.NodeID]bool)
|
||||
exitedLastWeek := make(map[storj.NodeID]bool)
|
||||
|
||||
now := time.Now()
|
||||
lastWeek := time.Now().AddDate(0, 0, -7)
|
||||
|
||||
testData := []struct {
|
||||
nodeID storj.NodeID
|
||||
initiatedAt time.Time
|
||||
completedAt time.Time
|
||||
finishedAt time.Time
|
||||
}{
|
||||
// exited today
|
||||
{testrand.NodeID(), now, now, now},
|
||||
// exited last week
|
||||
{testrand.NodeID(), lastWeek, lastWeek, lastWeek},
|
||||
// exiting today
|
||||
{testrand.NodeID(), now, now, time.Time{}},
|
||||
// exiting last week
|
||||
{testrand.NodeID(), lastWeek, lastWeek, time.Time{}},
|
||||
// not exiting
|
||||
{testrand.NodeID(), time.Time{}, time.Time{}, time.Time{}},
|
||||
}
|
||||
|
||||
for _, data := range testData {
|
||||
err := cache.UpdateAddress(ctx, &pb.Node{Id: data.nodeID}, overlay.NodeSelectionConfig{})
|
||||
require.NoError(t, err)
|
||||
|
||||
req := &overlay.ExitStatusRequest{
|
||||
NodeID: data.nodeID,
|
||||
ExitInitiatedAt: data.initiatedAt,
|
||||
ExitLoopCompletedAt: data.completedAt,
|
||||
ExitFinishedAt: data.finishedAt,
|
||||
}
|
||||
_, err = cache.UpdateExitStatus(ctx, req)
|
||||
require.NoError(t, err)
|
||||
|
||||
if !data.finishedAt.IsZero() {
|
||||
if data.finishedAt == now {
|
||||
exitedToday[data.nodeID] = true
|
||||
} else {
|
||||
exitedLastWeek[data.nodeID] = true
|
||||
}
|
||||
} else if !data.initiatedAt.IsZero() {
|
||||
if data.initiatedAt == now {
|
||||
exitingToday[data.nodeID] = true
|
||||
} else {
|
||||
exitingLastWeek[data.nodeID] = true
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// test GetGracefulExitIncompleteByTimeFrame
|
||||
ids, err := cache.GetGracefulExitIncompleteByTimeFrame(ctx, lastWeek.Add(-24*time.Hour), lastWeek.Add(24*time.Hour))
|
||||
require.NoError(t, err)
|
||||
require.Len(t, ids, 1)
|
||||
for _, id := range ids {
|
||||
require.True(t, exitingLastWeek[id])
|
||||
}
|
||||
ids, err = cache.GetGracefulExitIncompleteByTimeFrame(ctx, now.Add(-24*time.Hour), now.Add(24*time.Hour))
|
||||
require.NoError(t, err)
|
||||
require.Len(t, ids, 1)
|
||||
for _, id := range ids {
|
||||
require.True(t, exitingToday[id])
|
||||
}
|
||||
ids, err = cache.GetGracefulExitIncompleteByTimeFrame(ctx, lastWeek.Add(-24*time.Hour), now.Add(24*time.Hour))
|
||||
require.NoError(t, err)
|
||||
require.Len(t, ids, 2)
|
||||
for _, id := range ids {
|
||||
require.True(t, exitingLastWeek[id] || exitingToday[id])
|
||||
}
|
||||
|
||||
// test GetGracefulExitCompletedByTimeFrame
|
||||
ids, err = cache.GetGracefulExitCompletedByTimeFrame(ctx, lastWeek.Add(-24*time.Hour), lastWeek.Add(24*time.Hour))
|
||||
require.NoError(t, err)
|
||||
require.Len(t, ids, 1)
|
||||
for _, id := range ids {
|
||||
require.True(t, exitedLastWeek[id])
|
||||
}
|
||||
ids, err = cache.GetGracefulExitCompletedByTimeFrame(ctx, now.Add(-24*time.Hour), now.Add(24*time.Hour))
|
||||
require.NoError(t, err)
|
||||
require.Len(t, ids, 1)
|
||||
for _, id := range ids {
|
||||
require.True(t, exitedToday[id])
|
||||
}
|
||||
ids, err = cache.GetGracefulExitCompletedByTimeFrame(ctx, lastWeek.Add(-24*time.Hour), now.Add(24*time.Hour))
|
||||
require.NoError(t, err)
|
||||
require.Len(t, ids, 2)
|
||||
for _, id := range ids {
|
||||
require.True(t, exitedLastWeek[id] || exitedToday[id])
|
||||
}
|
||||
|
||||
})
|
||||
}
|
||||
|
||||
@ -77,7 +77,11 @@ type DB interface {
|
||||
GetExitingNodes(ctx context.Context) (exitingNodes storj.NodeIDList, err error)
|
||||
// GetExitingNodesLoopIncomplete returns exiting nodes who haven't completed the metainfo loop iteration.
|
||||
GetExitingNodesLoopIncomplete(ctx context.Context) (exitingNodes storj.NodeIDList, err error)
|
||||
|
||||
// GetGracefulExitCompletedByTimeFrame returns nodes who have completed graceful exit within a time window (time window is around graceful exit completion).
|
||||
GetGracefulExitCompletedByTimeFrame(ctx context.Context, begin, end time.Time) (exitedNodes storj.NodeIDList, err error)
|
||||
// GetGracefulExitIncompleteByTimeFrame returns nodes who have initiated, but not completed graceful exit within a time window (time window is around graceful exit initiation).
|
||||
GetGracefulExitIncompleteByTimeFrame(ctx context.Context, begin, end time.Time) (exitingNodes storj.NodeIDList, err error)
|
||||
// GetExitStatus returns a node's graceful exit status.
|
||||
GetExitStatus(ctx context.Context, nodeID storj.NodeID) (exitStatus *ExitStatus, err error)
|
||||
}
|
||||
|
||||
@ -161,6 +165,7 @@ type NodeDossier struct {
|
||||
Disqualified *time.Time
|
||||
PieceCount int64
|
||||
ExitStatus ExitStatus
|
||||
CreatedAt time.Time
|
||||
}
|
||||
|
||||
// NodeStats contains statistics about a node.
|
||||
|
||||
@ -404,6 +404,7 @@ func TestUpdateCheckIn(t *testing.T) {
|
||||
expectedNode.Reputation.LastContactSuccess = actualNode.Reputation.LastContactSuccess
|
||||
expectedNode.Reputation.LastContactFailure = actualNode.Reputation.LastContactFailure
|
||||
expectedNode.Version.Timestamp = actualNode.Version.Timestamp
|
||||
expectedNode.CreatedAt = actualNode.CreatedAt
|
||||
require.Equal(t, expectedNode, actualNode)
|
||||
|
||||
// confirm that we can update the address field
|
||||
|
||||
@ -47,7 +47,10 @@ func (db *gracefulexitDB) IncrementProgress(ctx context.Context, nodeID storj.No
|
||||
func (db *gracefulexitDB) GetProgress(ctx context.Context, nodeID storj.NodeID) (_ *gracefulexit.Progress, err error) {
|
||||
defer mon.Task()(&ctx)(&err)
|
||||
dbxProgress, err := db.db.Get_GracefulExitProgress_By_NodeId(ctx, dbx.GracefulExitProgress_NodeId(nodeID.Bytes()))
|
||||
if err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, gracefulexit.ErrNodeNotFound.Wrap(err)
|
||||
|
||||
} else if err != nil {
|
||||
return nil, Error.Wrap(err)
|
||||
}
|
||||
nID, err := storj.NodeIDFromBytes(dbxProgress.NodeId)
|
||||
|
||||
@ -884,6 +884,8 @@ func (cache *overlaycache) GetExitingNodesLoopIncomplete(ctx context.Context) (e
|
||||
}
|
||||
return exitingNodes, nil
|
||||
}
|
||||
|
||||
// GetExitStatus returns a node's graceful exit status.
|
||||
func (cache *overlaycache) GetExitStatus(ctx context.Context, nodeID storj.NodeID) (_ *overlay.ExitStatus, err error) {
|
||||
defer mon.Task()(&ctx)(&err)
|
||||
|
||||
@ -902,6 +904,67 @@ func (cache *overlaycache) GetExitStatus(ctx context.Context, nodeID storj.NodeI
|
||||
return exitStatus, Error.Wrap(err)
|
||||
}
|
||||
|
||||
// GetGracefulExitCompletedByTimeFrame returns nodes who have completed graceful exit within a time window (time window is around graceful exit completion).
|
||||
func (cache *overlaycache) GetGracefulExitCompletedByTimeFrame(ctx context.Context, begin, end time.Time) (exitedNodes storj.NodeIDList, err error) {
|
||||
defer mon.Task()(&ctx)(&err)
|
||||
|
||||
rows, err := cache.db.Query(cache.db.Rebind(`
|
||||
SELECT id FROM nodes
|
||||
WHERE exit_initiated_at IS NOT NULL
|
||||
AND exit_finished_at IS NOT NULL
|
||||
AND exit_finished_at >= ?
|
||||
AND exit_finished_at < ?
|
||||
`), begin, end,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
err = errs.Combine(err, rows.Close())
|
||||
}()
|
||||
|
||||
for rows.Next() {
|
||||
var id storj.NodeID
|
||||
err = rows.Scan(&id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
exitedNodes = append(exitedNodes, id)
|
||||
}
|
||||
return exitedNodes, rows.Err()
|
||||
}
|
||||
|
||||
// GetGracefulExitIncompleteByTimeFrame returns nodes who have initiated, but not completed graceful exit within a time window (time window is around graceful exit initiation).
|
||||
func (cache *overlaycache) GetGracefulExitIncompleteByTimeFrame(ctx context.Context, begin, end time.Time) (exitingNodes storj.NodeIDList, err error) {
|
||||
defer mon.Task()(&ctx)(&err)
|
||||
|
||||
rows, err := cache.db.Query(cache.db.Rebind(`
|
||||
SELECT id FROM nodes
|
||||
WHERE exit_initiated_at IS NOT NULL
|
||||
AND exit_finished_at IS NULL
|
||||
AND exit_initiated_at >= ?
|
||||
AND exit_initiated_at < ?
|
||||
`), begin, end,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
err = errs.Combine(err, rows.Close())
|
||||
}()
|
||||
|
||||
// TODO return more than just ID
|
||||
for rows.Next() {
|
||||
var id storj.NodeID
|
||||
err = rows.Scan(&id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
exitingNodes = append(exitingNodes, id)
|
||||
}
|
||||
return exitingNodes, rows.Err()
|
||||
}
|
||||
|
||||
// UpdateExitStatus is used to update a node's graceful exit status.
|
||||
func (cache *overlaycache) UpdateExitStatus(ctx context.Context, request *overlay.ExitStatusRequest) (stats *overlay.NodeStats, err error) {
|
||||
defer mon.Task()(&ctx)(&err)
|
||||
@ -993,6 +1056,7 @@ func convertDBNode(ctx context.Context, info *dbx.Node) (_ *overlay.NodeDossier,
|
||||
Disqualified: info.Disqualified,
|
||||
PieceCount: info.PieceCount,
|
||||
ExitStatus: exitStatus,
|
||||
CreatedAt: info.CreatedAt,
|
||||
}
|
||||
|
||||
return node, nil
|
||||
|
||||
Loading…
Reference in New Issue
Block a user