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

Storagenode add daily bandwidth usage query for SNO (#2348)

Browse Source
This commit is contained in:
Yaroslav Vorobiov 2019-07-02 12:53:40 +03:00 committed by Stefan Benten
parent 3337087a94
commit 16cd1fde87
3 changed files with 167 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
storagenode/console/bandwidth.go
View File

@ -4,11 +4,26 @@
package console
import (
"time"
"github.com/zeebo/errs"
"storj.io/storj/storagenode/bandwidth"
)
// Egress stores info about storage node egress usage
type Egress struct {
Repair int64 `json:"repair"`
Audit int64 `json:"audit"`
Usage int64 `json:"usage"`
}
// Ingress stores info about storage node ingress usage
type Ingress struct {
Repair int64 `json:"repair"`
Usage int64 `json:"usage"`
}
// BandwidthInfo stores all info about storage node bandwidth usage
type BandwidthInfo struct {
Egress Egress `json:"egress"`
@ -17,6 +32,15 @@ type BandwidthInfo struct {
Remaining int64 `json:"remaining"`
}
// BandwidthUsed stores bandwidth usage information
// over the period of time
type BandwidthUsed struct {
Egress Egress `json:"egress"`
Ingress Ingress `json:"ingress"`
From, To time.Time
}
// FromUsage used to create BandwidthInfo instance from Usage object
func FromUsage(usage *bandwidth.Usage, allocatedBandwidth int64) (*BandwidthInfo, error) {
if usage == nil {
@ -39,16 +63,3 @@ func FromUsage(usage *bandwidth.Usage, allocatedBandwidth int64) (*BandwidthInfo
Used: used,
}, nil
}
// Egress stores info about storage node egress usage
type Egress struct {
Repair int64 `json:"repair"`
Audit int64 `json:"audit"`
Usage int64 `json:"usage"`
}
// Ingress stores info about storage node ingress usage
type Ingress struct {
Repair int64 `json:"repair"`
Usage int64 `json:"usage"`
}

36
storagenode/console/service.go
View File

@ -9,7 +9,7 @@ import (
"github.com/zeebo/errs"
"go.uber.org/zap"
monkit "gopkg.in/spacemonkeygo/monkit.v2"
"gopkg.in/spacemonkeygo/monkit.v2"
"storj.io/storj/internal/memory"
"storj.io/storj/internal/version"
@ -29,7 +29,15 @@ var (
// DB exposes methods for managing SNO dashboard related data.
type DB interface {
// GetSatelliteIDs returns list of satelliteIDs that storagenode has interacted with
// at least once
GetSatelliteIDs(ctx context.Context, from, to time.Time) (storj.NodeIDList, error)
// GetDailyBandwidthUsed returns slice of daily bandwidth usage for provided time range,
// sorted in ascending order
GetDailyTotalBandwidthUsed(ctx context.Context, from, to time.Time) ([]BandwidthUsed, error)
// GetDailyBandwidthUsed returns slice of daily bandwidth usage for provided time range,
// sorted in ascending order for particular satellite
GetDailyBandwidthUsed(ctx context.Context, satelliteID storj.NodeID, from, to time.Time) ([]BandwidthUsed, error)
}
// Service is handling storage node operator related logic
@ -105,6 +113,32 @@ func (s *Service) GetUsedBandwidthTotal(ctx context.Context) (_ *BandwidthInfo,
return FromUsage(usage, s.allocatedBandwidth.Int64())
}
// GetDailyTotalBandwidthUsed returns slice of daily bandwidth usage for provided time range,
// sorted in ascending order
func (s *Service) GetDailyTotalBandwidthUsed(ctx context.Context, from, to time.Time) (_ []BandwidthUsed, err error) {
defer mon.Task()(&ctx)(&err)
usage, err := s.consoleDB.GetDailyTotalBandwidthUsed(ctx, from, to)
if err != nil {
return nil, err
}
return usage, nil
}
// GetDailyBandwidthUsed returns slice of daily bandwidth usage for provided time range,
// sorted in ascending order for particular satellite
func (s *Service) GetDailyBandwidthUsed(ctx context.Context, satelliteID storj.NodeID, from, to time.Time) (_ []BandwidthUsed, err error) {
defer mon.Task()(&ctx)(&err)
usage, err := s.consoleDB.GetDailyBandwidthUsed(ctx, satelliteID, from, to)
if err != nil {
return nil, err
}
return usage, nil
}
// GetBandwidthBySatellite returns all info about storage node bandwidth usage by satellite
func (s *Service) GetBandwidthBySatellite(ctx context.Context, satelliteID storj.NodeID) (_ *BandwidthInfo, err error) {
defer mon.Task()(&ctx)(&err)

108
storagenode/storagenodedb/consoledb.go
View File

@ -6,10 +6,12 @@ package storagenodedb
import (
"context"
"database/sql"
"strings"
"time"
"github.com/zeebo/errs"
"storj.io/storj/pkg/pb"
"storj.io/storj/pkg/storj"
"storj.io/storj/storagenode/console"
)
@ -22,6 +24,8 @@ func (db *InfoDB) Console() console.DB { return &consoledb{db} }
// Console returns console.DB
func (db *DB) Console() console.DB { return db.info.Console() }
// GetSatelliteIDs returns list of satelliteIDs that storagenode has interacted with
// at least once
func (db *consoledb) GetSatelliteIDs(ctx context.Context, from, to time.Time) (_ storj.NodeIDList, err error) {
defer mon.Task()(&ctx)(&err)
defer db.locked()()
@ -54,3 +58,107 @@ func (db *consoledb) GetSatelliteIDs(ctx context.Context, from, to time.Time) (_
return satellites, nil
}
// GetDailyBandwidthUsed returns slice of daily bandwidth usage for provided time range,
// sorted in ascending order
func (db *consoledb) GetDailyTotalBandwidthUsed(ctx context.Context, from, to time.Time) (_ []console.BandwidthUsed, err error) {
defer mon.Task()(&ctx)(&err)
since, _ := getDateEdges(from.UTC())
_, before := getDateEdges(to.UTC())
return db.getDailyBandwidthUsed(ctx,
"WHERE ? <= created_at AND created_at <= ?",
since, before)
}
// GetDailyBandwidthUsed returns slice of daily bandwidth usage for provided time range,
// sorted in ascending order for particular satellite
func (db *consoledb) GetDailyBandwidthUsed(ctx context.Context, satelliteID storj.NodeID, from, to time.Time) (_ []console.BandwidthUsed, err error) {
defer mon.Task()(&ctx)(&err)
since, _ := getDateEdges(from.UTC())
_, before := getDateEdges(to.UTC())
return db.getDailyBandwidthUsed(ctx,
"WHERE satellite_id = ? AND ? <= created_at AND created_at <= ?",
satelliteID, since, before)
}
// getDailyBandwidthUsed returns slice of grouped by date bandwidth usage
// sorted in ascending order and applied condition if any
func (db *consoledb) getDailyBandwidthUsed(ctx context.Context, cond string, args ...interface{}) (_ []console.BandwidthUsed, err error) {
defer mon.Task()(&ctx)(&err)
defer db.locked()()
qb := strings.Builder{}
qb.WriteString("SELECT action, SUM(amount), created_at ")
qb.WriteString("FROM bandwidth_usage ")
if cond != "" {
qb.WriteString(cond + " ")
}
qb.WriteString("GROUP BY DATE(created_at), action ")
qb.WriteString("ORDER BY created_at ASC")
rows, err := db.db.QueryContext(ctx, db.Rebind(qb.String()), args)
if err != nil {
return nil, err
}
defer func() {
err = errs.Combine(err, rows.Close())
}()
var dates []time.Time
dailyBandwidth := make(map[time.Time]*console.BandwidthUsed, 0)
for rows.Next() {
var action int32
var amount int64
var createdAt time.Time
err = rows.Scan(&action, &amount, &createdAt)
if err != nil {
return nil, err
}
from, to := getDateEdges(createdAt)
bandwidthUsed, ok := dailyBandwidth[from]
if !ok {
bandwidthUsed = &console.BandwidthUsed{
From: from,
To: to,
}
dates = append(dates, from)
dailyBandwidth[from] = bandwidthUsed
}
switch pb.PieceAction(action) {
case pb.PieceAction_GET:
bandwidthUsed.Egress.Usage = amount
case pb.PieceAction_GET_AUDIT:
bandwidthUsed.Egress.Audit = amount
case pb.PieceAction_GET_REPAIR:
bandwidthUsed.Egress.Repair = amount
case pb.PieceAction_PUT:
bandwidthUsed.Ingress.Usage = amount
case pb.PieceAction_PUT_REPAIR:
bandwidthUsed.Ingress.Repair = amount
}
}
var bandwidthUsedList []console.BandwidthUsed
for _, date := range dates {
bandwidthUsedList = append(bandwidthUsedList, *dailyBandwidth[date])
}
return bandwidthUsedList, nil
}
// getDateEdges returns start and end of the provided day
func getDateEdges(t time.Time) (time.Time, time.Time) {
return time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, time.UTC),
time.Date(t.Year(), t.Month(), t.Day()+1, 0, 0, 0, -1, time.UTC)
}