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

multinode/bandwidth: added monthly bandwidth summaries

Browse Source 
montly bandwidth summaries for single/all nodes, single/all satellites added.

Change-Id: Ic384886c10622df74a4bd0645e2d7f2a85477644
This commit is contained in:
Qweder93 2021-06-02 15:13:42 +03:00 committed by Nikolai Siedov
parent f3a52d1da5
commit 5d70b6abef
11 changed files with 2284 additions and 165 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

93
multinode/bandwidth/bandwidth.go Normal file
View File

@ -0,0 +1,93 @@
// Copyright (C) 2021 Storj Labs, Inc.
// See LICENSE for copying information.
package bandwidth
import (
"sort"
"time"
)
// 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"`
}
// UsageRollup contains rolluped bandwidth usage.
type UsageRollup struct {
Egress Egress `json:"egress"`
Ingress Ingress `json:"ingress"`
Delete int64 `json:"delete"`
IntervalStart time.Time `json:"intervalStart"`
}
// Monthly contains all bandwidth, ingress, egress monthly data.
type Monthly struct {
BandwidthDaily []UsageRollup `json:"bandwidthDaily"`
BandwidthSummary int64 `json:"bandwidthSummary"`
EgressSummary int64 `json:"egressSummary"`
IngressSummary int64 `json:"ingressSummary"`
}
// UsageRollupDailyCache caches storage usage stamps by interval date.
type UsageRollupDailyCache map[time.Time]UsageRollup
// Sorted returns usage rollup slice sorted by interval start.
func (cache *UsageRollupDailyCache) Sorted() []UsageRollup {
var usageRollup []UsageRollup
for _, stamp := range *cache {
usageRollup = append(usageRollup, stamp)
}
sort.Slice(usageRollup, func(i, j int) bool {
return usageRollup[i].IntervalStart.Before(usageRollup[j].IntervalStart)
})
return usageRollup
}
// Add adds usage rollup to cache aggregating bandwidth data by date.
func (cache *UsageRollupDailyCache) Add(rollup UsageRollup) {
year, month, day := rollup.IntervalStart.UTC().Date()
intervalStart := time.Date(year, month, day, 0, 0, 0, 0, time.UTC)
cached := *cache
cacheStamp, ok := cached[intervalStart]
if ok {
cached[intervalStart] = UsageRollup{
Egress: Egress{
Repair: cacheStamp.Egress.Repair + rollup.Egress.Repair,
Audit: cacheStamp.Egress.Audit + rollup.Egress.Audit,
Usage: cacheStamp.Egress.Usage + rollup.Egress.Usage,
},
Ingress: Ingress{
Repair: cacheStamp.Ingress.Repair + rollup.Ingress.Repair,
Usage: cacheStamp.Ingress.Usage + rollup.Ingress.Usage,
},
Delete: cacheStamp.Delete + rollup.Delete,
IntervalStart: intervalStart,
}
} else {
cached[intervalStart] = UsageRollup{
Egress: Egress{
Repair: rollup.Egress.Repair,
Audit: rollup.Egress.Audit,
Usage: rollup.Egress.Usage,
},
Ingress: Ingress{
Repair: rollup.Ingress.Repair,
Usage: rollup.Ingress.Usage,
},
Delete: rollup.Delete,
IntervalStart: intervalStart,
}
}
*cache = cached
}

103
multinode/bandwidth/bandwidth_test.go Normal file
View File

@ -0,0 +1,103 @@
// Copyright (C) 2021 Storj Labs, Inc.
// See LICENSE for copying information.
package bandwidth_test
import (
"testing"
"time"
"github.com/stretchr/testify/require"
"storj.io/storj/multinode/bandwidth"
)
func TestUsageRolloutDailyCache(t *testing.T) {
newTimestamp := func(month time.Month, day int) time.Time {
return time.Date(2021, month, day, 0, 0, 0, 0, time.UTC)
}
testData := []struct {
Date time.Time
Egress []bandwidth.Egress
Ingress []bandwidth.Ingress
Delete []int64
}{
{
Date: newTimestamp(time.May, 2),
Ingress: []bandwidth.Ingress{{Repair: 1, Usage: 0}, {Repair: 10, Usage: 20}},
Egress: []bandwidth.Egress{{Repair: 10, Audit: 20, Usage: 30}, {Repair: 10, Audit: 20, Usage: 30}},
Delete: []int64{10, 20, 30},
},
{
Date: newTimestamp(time.May, 3),
Ingress: []bandwidth.Ingress{{Repair: 1, Usage: 0}, {Repair: 10, Usage: 20}},
Egress: []bandwidth.Egress{{Repair: 101, Audit: 201, Usage: 301}, {Repair: 101, Audit: 201, Usage: 301}},
Delete: []int64{101, 201, 301},
},
{
Date: newTimestamp(time.May, 4),
Ingress: []bandwidth.Ingress{{Repair: 12, Usage: 20}, {Repair: 120, Usage: 220}},
Egress: []bandwidth.Egress{{Repair: 310, Audit: 320, Usage: 330}, {Repair: 100, Audit: 200, Usage: 300}},
Delete: []int64{100, 200, 300},
},
{
Date: newTimestamp(time.May, 1),
Ingress: []bandwidth.Ingress{{Repair: 123, Usage: 123}, {Repair: 123, Usage: 123}},
Egress: []bandwidth.Egress{{Repair: 20, Audit: 20, Usage: 20}, {Repair: 30, Audit: 30, Usage: 30}},
Delete: []int64{2, 3, 4},
},
}
expected := []bandwidth.UsageRollup{
{
IntervalStart: newTimestamp(time.May, 1),
Ingress: bandwidth.Ingress{Repair: 246, Usage: 246},
Egress: bandwidth.Egress{Repair: 50, Audit: 50, Usage: 50},
Delete: 9,
},
{
IntervalStart: newTimestamp(time.May, 2),
Ingress: bandwidth.Ingress{Repair: 11, Usage: 20},
Egress: bandwidth.Egress{Repair: 20, Audit: 40, Usage: 60},
Delete: 60,
},
{
IntervalStart: newTimestamp(time.May, 3),
Ingress: bandwidth.Ingress{Repair: 11, Usage: 20},
Egress: bandwidth.Egress{Repair: 202, Audit: 402, Usage: 602},
Delete: 603,
},
{
IntervalStart: newTimestamp(time.May, 4),
Ingress: bandwidth.Ingress{Repair: 132, Usage: 240},
Egress: bandwidth.Egress{Repair: 410, Audit: 520, Usage: 630},
Delete: 600,
},
}
cache := make(bandwidth.UsageRollupDailyCache)
for _, entry := range testData {
_, month, day := entry.Date.Date()
for _, egr := range entry.Egress {
cache.Add(bandwidth.UsageRollup{
Egress: egr,
IntervalStart: newTimestamp(month, day),
})
}
for _, ing := range entry.Ingress {
cache.Add(bandwidth.UsageRollup{
Ingress: ing,
IntervalStart: newTimestamp(month, day),
})
}
for _, del := range entry.Delete {
cache.Add(bandwidth.UsageRollup{
Delete: del,
IntervalStart: newTimestamp(month, day),
})
}
}
stamps := cache.Sorted()
require.Equal(t, expected, stamps)
}

289
multinode/bandwidth/service.go Normal file
View File

@ -0,0 +1,289 @@
// Copyright (C) 2021 Storj Labs, Inc.
// See LICENSE for copying information.
package bandwidth
import (
"context"
"github.com/spacemonkeygo/monkit/v3"
"github.com/zeebo/errs"
"go.uber.org/zap"
"storj.io/common/rpc"
"storj.io/common/storj"
"storj.io/storj/multinode/nodes"
"storj.io/storj/private/multinodepb"
)
var (
mon = monkit.Package()
// Error is an error class for bandwidth service error.
Error = errs.Class("bandwidth")
)
// Service exposes bandwidth related logic.
//
// architecture: Service
type Service struct {
log *zap.Logger
dialer rpc.Dialer
nodes *nodes.Service
}
// NewService creates new instance of Service.
func NewService(log *zap.Logger, dialer rpc.Dialer, nodes *nodes.Service) *Service {
return &Service{
log: log,
dialer: dialer,
nodes: nodes,
}
}
// Monthly returns monthly bandwidth summary.
func (service *Service) Monthly(ctx context.Context) (_ Monthly, err error) {
defer mon.Task()(&ctx)(&err)
var totalMonthly Monthly
nodes, err := service.nodes.List(ctx)
if err != nil {
return Monthly{}, Error.Wrap(err)
}
cache := make(UsageRollupDailyCache)
for _, node := range nodes {
monthly, err := service.getMonthly(ctx, node)
if err != nil {
return Monthly{}, Error.Wrap(err)
}
totalMonthly.IngressSummary += monthly.IngressSummary
totalMonthly.EgressSummary += monthly.EgressSummary
totalMonthly.BandwidthSummary += monthly.BandwidthSummary
for _, rollup := range monthly.BandwidthDaily {
cache.Add(rollup)
}
}
totalMonthly.BandwidthDaily = cache.Sorted()
return totalMonthly, nil
}
// MonthlyNode returns monthly bandwidth summary for single node.
func (service *Service) MonthlyNode(ctx context.Context, nodeID storj.NodeID) (_ Monthly, err error) {
defer mon.Task()(&ctx)(&err)
node, err := service.nodes.Get(ctx, nodeID)
if err != nil {
return Monthly{}, Error.Wrap(err)
}
monthly, err := service.getMonthly(ctx, node)
if err != nil {
return Monthly{}, Error.Wrap(err)
}
return monthly, nil
}
// MonthlySatellite returns monthly bandwidth summary for specific satellite.
func (service *Service) MonthlySatellite(ctx context.Context, satelliteID storj.NodeID) (_ Monthly, err error) {
defer mon.Task()(&ctx)(&err)
var totalMonthly Monthly
nodes, err := service.nodes.List(ctx)
if err != nil {
return Monthly{}, Error.Wrap(err)
}
cache := make(UsageRollupDailyCache)
for _, node := range nodes {
monthly, err := service.getMonthlySatellite(ctx, node, satelliteID)
if err != nil {
return Monthly{}, Error.Wrap(err)
}
totalMonthly.IngressSummary += monthly.IngressSummary
totalMonthly.EgressSummary += monthly.EgressSummary
totalMonthly.BandwidthSummary += monthly.BandwidthSummary
for _, rollup := range monthly.BandwidthDaily {
cache.Add(rollup)
}
}
totalMonthly.BandwidthDaily = cache.Sorted()
return totalMonthly, nil
}
// MonthlySatelliteNode returns monthly bandwidth summary for single node and specific satellites.
func (service *Service) MonthlySatelliteNode(ctx context.Context, satelliteID, nodeID storj.NodeID) (_ Monthly, err error) {
defer mon.Task()(&ctx)(&err)
node, err := service.nodes.Get(ctx, nodeID)
if err != nil {
return Monthly{}, Error.Wrap(err)
}
monthly, err := service.getMonthlySatellite(ctx, node, satelliteID)
if err != nil {
return Monthly{}, Error.Wrap(err)
}
return monthly, nil
}
// getMonthlySatellite returns monthly bandwidth summary for single node and specific satellite.
func (service *Service) getMonthlySatellite(ctx context.Context, node nodes.Node, satelliteID storj.NodeID) (_ Monthly, err error) {
defer mon.Task()(&ctx)(&err)
conn, err := service.dialer.DialNodeURL(ctx, storj.NodeURL{
ID: node.ID,
Address: node.PublicAddress,
})
if err != nil {
return Monthly{}, Error.Wrap(err)
}
defer func() {
err = errs.Combine(err, conn.Close())
}()
bandwidthClient := multinodepb.NewDRPCBandwidthClient(conn)
header := &multinodepb.RequestHeader{
ApiKey: node.APISecret,
}
ingress, err := bandwidthClient.IngressSummarySatellite(ctx, &multinodepb.IngressSummarySatelliteRequest{
Header: header,
SatelliteId: satelliteID,
})
if err != nil {
return Monthly{}, Error.Wrap(err)
}
egress, err := bandwidthClient.EgressSummarySatellite(ctx, &multinodepb.EgressSummarySatelliteRequest{
Header: header,
SatelliteId: satelliteID,
})
if err != nil {
return Monthly{}, Error.Wrap(err)
}
bandwidth, err := bandwidthClient.BandwidthSummarySatellite(ctx, &multinodepb.BandwidthSummarySatelliteRequest{
Header: header,
SatelliteId: satelliteID,
})
if err != nil {
return Monthly{}, Error.Wrap(err)
}
usageRollup, err := bandwidthClient.DailySatellite(ctx, &multinodepb.DailySatelliteRequest{
Header: header,
SatelliteId: satelliteID,
})
if err != nil {
return Monthly{}, Error.Wrap(err)
}
var rollups []UsageRollup
for _, r := range usageRollup.UsageRollup {
rollups = append(rollups, UsageRollup{
Egress: Egress{
Repair: r.Egress.Repair,
Audit: r.Egress.Audit,
Usage: r.Egress.Usage,
},
Ingress: Ingress{
Repair: r.Ingress.Repaid,
Usage: r.Ingress.Usage,
},
Delete: r.Delete,
IntervalStart: r.IntervalStart,
})
}
return Monthly{
BandwidthDaily: rollups,
BandwidthSummary: bandwidth.Summary,
EgressSummary: egress.Summary,
IngressSummary: ingress.Summary,
}, nil
}
// getMonthly returns monthly bandwidth summary for single node.
func (service *Service) getMonthly(ctx context.Context, node nodes.Node) (_ Monthly, err error) {
defer mon.Task()(&ctx)(&err)
conn, err := service.dialer.DialNodeURL(ctx, storj.NodeURL{
ID: node.ID,
Address: node.PublicAddress,
})
if err != nil {
return Monthly{}, Error.Wrap(err)
}
defer func() {
err = errs.Combine(err, conn.Close())
}()
bandwidthClient := multinodepb.NewDRPCBandwidthClient(conn)
header := &multinodepb.RequestHeader{
ApiKey: node.APISecret,
}
ingress, err := bandwidthClient.IngressSummary(ctx, &multinodepb.IngressSummaryRequest{
Header: header,
})
if err != nil {
return Monthly{}, Error.Wrap(err)
}
egress, err := bandwidthClient.EgressSummary(ctx, &multinodepb.EgressSummaryRequest{
Header: header,
})
if err != nil {
return Monthly{}, Error.Wrap(err)
}
bandwidth, err := bandwidthClient.BandwidthSummary(ctx, &multinodepb.BandwidthSummaryRequest{
Header: header,
})
if err != nil {
return Monthly{}, Error.Wrap(err)
}
usageRollup, err := bandwidthClient.Daily(ctx, &multinodepb.DailyRequest{
Header: header,
})
if err != nil {
return Monthly{}, Error.Wrap(err)
}
var rollups []UsageRollup
for _, r := range usageRollup.UsageRollup {
rollups = append(rollups, UsageRollup{
Egress: Egress{
Repair: r.Egress.Repair,
Audit: r.Egress.Audit,
Usage: r.Egress.Usage,
},
Ingress: Ingress{
Repair: r.Ingress.Repaid,
Usage: r.Ingress.Usage,
},
Delete: r.Delete,
IntervalStart: r.IntervalStart,
})
}
return Monthly{
BandwidthDaily: rollups,
BandwidthSummary: bandwidth.Summary,
EgressSummary: egress.Summary,
IngressSummary: ingress.Summary,
}, nil
}

186
multinode/console/controllers/bandwidth.go Normal file
View File

@ -0,0 +1,186 @@
// Copyright (C) 2021 Storj Labs, Inc.
// See LICENSE for copying information.
package controllers
import (
"encoding/json"
"net/http"
"github.com/gorilla/mux"
"github.com/zeebo/errs"
"go.uber.org/zap"
"storj.io/common/storj"
"storj.io/storj/multinode/bandwidth"
)
var (
// ErrBandwidth is an internal error type for bandwidth web api controller.
ErrBandwidth = errs.Class("bandwidth web api controller")
)
// Bandwidth is a web api controller.
type Bandwidth struct {
log *zap.Logger
service *bandwidth.Service
}
// NewBandwidth is a constructor for Bandwidth.
func NewBandwidth(log *zap.Logger, service *bandwidth.Service) *Bandwidth {
return &Bandwidth{
log: log,
service: service,
}
}
// Monthly handles all satellites all nodes bandwidth monthly.
func (controller *Bandwidth) Monthly(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
var err error
defer mon.Task()(&ctx)(&err)
w.Header().Add("Content-Type", "application/json")
monthly, err := controller.service.Monthly(ctx)
if err != nil {
controller.log.Error("get bandwidth monthly error", zap.Error(err))
controller.serveError(w, http.StatusInternalServerError, ErrBandwidth.Wrap(err))
return
}
if err = json.NewEncoder(w).Encode(monthly); err != nil {
controller.log.Error("failed to write json response", zap.Error(err))
return
}
}
// MonthlyNode handles all satellites single node bandwidth monthly.
func (controller *Bandwidth) MonthlyNode(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
var err error
defer mon.Task()(&ctx)(&err)
w.Header().Add("Content-Type", "application/json")
segmentParams := mux.Vars(r)
id, ok := segmentParams["nodeID"]
if !ok {
controller.serveError(w, http.StatusBadRequest, ErrPayouts.New("couldn't receive route variable nodeID"))
return
}
nodeID, err := storj.NodeIDFromString(id)
if err != nil {
controller.serveError(w, http.StatusBadRequest, ErrPayouts.Wrap(err))
return
}
monthly, err := controller.service.MonthlyNode(ctx, nodeID)
if err != nil {
controller.log.Error("get bandwidth monthly for specific node error", zap.Error(err))
controller.serveError(w, http.StatusInternalServerError, ErrBandwidth.Wrap(err))
return
}
if err = json.NewEncoder(w).Encode(monthly); err != nil {
controller.log.Error("failed to write json response", zap.Error(err))
return
}
}
// MonthlySatellite handles specific satellite all nodes bandwidth monthly.
func (controller *Bandwidth) MonthlySatellite(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
var err error
defer mon.Task()(&ctx)(&err)
w.Header().Add("Content-Type", "application/json")
segmentParams := mux.Vars(r)
id, ok := segmentParams["id"]
if !ok {
controller.serveError(w, http.StatusBadRequest, ErrPayouts.New("couldn't receive route variable satellite id"))
return
}
satelliteID, err := storj.NodeIDFromString(id)
if err != nil {
controller.serveError(w, http.StatusBadRequest, ErrPayouts.Wrap(err))
return
}
monthly, err := controller.service.MonthlySatellite(ctx, satelliteID)
if err != nil {
controller.log.Error("get bandwidth monthly for specific satellite error", zap.Error(err))
controller.serveError(w, http.StatusInternalServerError, ErrBandwidth.Wrap(err))
return
}
if err = json.NewEncoder(w).Encode(monthly); err != nil {
controller.log.Error("failed to write json response", zap.Error(err))
return
}
}
// MonthlySatelliteNode handles specific satellite single node bandwidth monthly.
func (controller *Bandwidth) MonthlySatelliteNode(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
var err error
defer mon.Task()(&ctx)(&err)
w.Header().Add("Content-Type", "application/json")
segmentParams := mux.Vars(r)
id, ok := segmentParams["id"]
if !ok {
controller.serveError(w, http.StatusBadRequest, ErrPayouts.New("couldn't receive route variable satellite id"))
return
}
satelliteID, err := storj.NodeIDFromString(id)
if err != nil {
controller.serveError(w, http.StatusBadRequest, ErrPayouts.Wrap(err))
return
}
node, ok := segmentParams["nodeID"]
if !ok {
controller.serveError(w, http.StatusBadRequest, ErrPayouts.New("couldn't receive route variable satellite id"))
return
}
nodeID, err := storj.NodeIDFromString(node)
if err != nil {
controller.serveError(w, http.StatusBadRequest, ErrPayouts.Wrap(err))
return
}
monthly, err := controller.service.MonthlySatelliteNode(ctx, satelliteID, nodeID)
if err != nil {
controller.log.Error("get bandwidth monthly for specific satellite and node error", zap.Error(err))
controller.serveError(w, http.StatusInternalServerError, ErrBandwidth.Wrap(err))
return
}
if err = json.NewEncoder(w).Encode(monthly); err != nil {
controller.log.Error("failed to write json response", zap.Error(err))
return
}
}
// serveError set http statuses and send json error.
func (controller *Bandwidth) serveError(w http.ResponseWriter, status int, err error) {
w.WriteHeader(status)
var response struct {
Error string `json:"error"`
}
response.Error = err.Error()
err = json.NewEncoder(w).Encode(response)
if err != nil {
controller.log.Error("failed to write json error response", zap.Error(err))
}
}

11
multinode/console/server/server.go
View File

@ -15,6 +15,7 @@ import (
"go.uber.org/zap"
"golang.org/x/sync/errgroup"
"storj.io/storj/multinode/bandwidth"
"storj.io/storj/multinode/console/controllers"
"storj.io/storj/multinode/nodes"
"storj.io/storj/multinode/operators"
@ -39,6 +40,7 @@ type Services struct {
Payouts *payouts.Service
Operators *operators.Service
Storage *storage.Service
Bandwidth *bandwidth.Service
}
// Server represents Multinode Dashboard http server.
@ -53,6 +55,7 @@ type Server struct {
nodes *nodes.Service
payouts *payouts.Service
operators *operators.Service
bandwidth *bandwidth.Service
storage *storage.Service
index *template.Template
@ -68,6 +71,7 @@ func NewServer(log *zap.Logger, listener net.Listener, config Config, services S
operators: services.Operators,
payouts: services.Payouts,
storage: services.Storage,
bandwidth: services.Bandwidth,
}
router := mux.NewRouter()
@ -90,6 +94,13 @@ func NewServer(log *zap.Logger, listener net.Listener, config Config, services S
operatorsRouter := apiRouter.PathPrefix("/operators").Subrouter()
operatorsRouter.HandleFunc("", operatorsController.ListPaginated).Methods(http.MethodGet)
bandwidthController := controllers.NewBandwidth(server.log, server.bandwidth)
bandwidthRouter := apiRouter.PathPrefix("/bandwidth").Subrouter()
bandwidthRouter.HandleFunc("/", bandwidthController.Monthly).Methods(http.MethodGet)
bandwidthRouter.HandleFunc("/{nodeID}", bandwidthController.MonthlyNode).Methods(http.MethodGet)
bandwidthRouter.HandleFunc("/satellites/{id}", bandwidthController.MonthlySatellite).Methods(http.MethodGet)
bandwidthRouter.HandleFunc("/satellites/{id}/{nodeID}", bandwidthController.MonthlySatelliteNode).Methods(http.MethodGet)
payoutsController := controllers.NewPayouts(server.log, server.payouts)
payoutsRouter := apiRouter.PathPrefix("/payouts").Subrouter()
payoutsRouter.HandleFunc("/summaries", payoutsController.Summary).Methods(http.MethodGet)

12
multinode/nodes/service.go
View File

@ -47,6 +47,18 @@ func (service *Service) Add(ctx context.Context, id storj.NodeID, apiSecret []by
return Error.Wrap(service.nodes.Add(ctx, id, apiSecret, publicAddress))
}
// List returns list of all nodes.
func (service *Service) List(ctx context.Context) (_ []Node, err error) {
defer mon.Task()(&ctx)(&err)
nodes, err := service.nodes.List(ctx)
if err != nil {
return nil, Error.Wrap(err)
}
return nodes, nil
}
// UpdateName will update name of the specified node.
func (service *Service) UpdateName(ctx context.Context, id storj.NodeID, name string) (err error) {
defer mon.Task()(&ctx)(&err)

15
multinode/peer.go
View File

@ -15,6 +15,7 @@ import (
"storj.io/common/peertls/tlsopts"
"storj.io/common/rpc"
"storj.io/private/debug"
"storj.io/storj/multinode/bandwidth"
"storj.io/storj/multinode/console/server"
"storj.io/storj/multinode/nodes"
"storj.io/storj/multinode/operators"
@ -64,6 +65,11 @@ type Peer struct {
Service *nodes.Service
}
// contains logic of bandwidth domain.
Bandwidth struct {
Service *bandwidth.Service
}
// exposes operators related logic.
Operators struct {
Service *operators.Service
@ -116,6 +122,14 @@ func New(log *zap.Logger, full *identity.FullIdentity, config Config, db DB) (_
)
}
{ // bandwidth setup
peer.Bandwidth.Service = bandwidth.NewService(
peer.Log.Named("bandwidth:service"),
peer.Dialer,
peer.Nodes.Service,
)
}
{ // operators setup
peer.Operators.Service = operators.NewService(
peer.Log.Named("operators:service"),
@ -155,6 +169,7 @@ func New(log *zap.Logger, full *identity.FullIdentity, config Config, db DB) (_
Payouts: peer.Payouts.Service,
Operators: peer.Operators.Service,
Storage: peer.Storage.Service,
Bandwidth: peer.Bandwidth.Service,
},
)
if err != nil {

1145
private/multinodepb/multinode.pb.go
View File

File diff suppressed because it is too large Load Diff

94
private/multinodepb/multinode.proto
View File

@ -60,6 +60,14 @@ message StorageUsageSatelliteResponse {
service Bandwidth {
rpc MonthSummary(BandwidthMonthSummaryRequest) returns (BandwidthMonthSummaryResponse);
rpc BandwidthSummarySatellite(BandwidthSummarySatelliteRequest) returns (BandwidthSummarySatelliteResponse);
rpc BandwidthSummary(BandwidthSummaryRequest) returns (BandwidthSummaryResponse);
rpc EgressSummarySatellite(EgressSummarySatelliteRequest) returns (EgressSummarySatelliteResponse);
rpc EgressSummary(EgressSummaryRequest) returns (EgressSummaryResponse);
rpc IngressSummarySatellite(IngressSummarySatelliteRequest) returns (IngressSummarySatelliteResponse);
rpc IngressSummary(IngressSummaryRequest) returns (IngressSummaryResponse);
rpc DailySatellite(DailySatelliteRequest) returns (DailySatelliteResponse);
rpc Daily(DailyRequest) returns (DailyResponse);
}
message BandwidthMonthSummaryRequest {
@ -70,6 +78,92 @@ message BandwidthMonthSummaryResponse {
int64 used = 1;
}
message BandwidthSummarySatelliteRequest {
RequestHeader header = 1;
bytes satellite_id = 2 [(gogoproto.customtype) = "NodeID", (gogoproto.nullable) = false];
}
message BandwidthSummarySatelliteResponse {
int64 summary = 1;
}
message BandwidthSummaryRequest {
RequestHeader header = 1;
}
message BandwidthSummaryResponse {
int64 summary = 1;
}
message EgressSummarySatelliteRequest {
RequestHeader header = 1;
bytes satellite_id = 2 [(gogoproto.customtype) = "NodeID", (gogoproto.nullable) = false];
}
message EgressSummarySatelliteResponse {
int64 summary = 1;
}
message EgressSummaryRequest {
RequestHeader header = 1;
}
message EgressSummaryResponse {
int64 summary = 1;
}
message IngressSummarySatelliteRequest {
RequestHeader header = 1;
bytes satellite_id = 2 [(gogoproto.customtype) = "NodeID", (gogoproto.nullable) = false];
}
message IngressSummarySatelliteResponse {
int64 summary = 1;
}
message IngressSummaryRequest {
RequestHeader header = 1;
}
message IngressSummaryResponse {
int64 summary = 1;
}
message DailySatelliteRequest {
RequestHeader header = 1;
bytes satellite_id = 2 [(gogoproto.customtype) = "NodeID", (gogoproto.nullable) = false];
}
message DailySatelliteResponse {
repeated UsageRollup usage_rollup = 1;
}
message DailyRequest {
RequestHeader header = 1;
}
message DailyResponse {
repeated UsageRollup usage_rollup = 1;
}
message UsageRollup {
Egress egress = 1;
Ingress ingress = 2;
int64 delete = 3;
google.protobuf.Timestamp interval_start = 4 [(gogoproto.stdtime) = true, (gogoproto.nullable) = false];
}
message Egress {
int64 repair = 1;
int64 audit = 2;
int64 usage = 3;
}
message Ingress {
int64 repaid = 1;
int64 usage = 2;
}
service Node {
rpc Version(VersionRequest) returns (VersionResponse);
rpc LastContact(LastContactRequest) returns (LastContactResponse);

322
private/multinodepb/multinode_drpc.pb.go
View File

@ -198,6 +198,14 @@ type DRPCBandwidthClient interface {
DRPCConn() drpc.Conn
MonthSummary(ctx context.Context, in *BandwidthMonthSummaryRequest) (*BandwidthMonthSummaryResponse, error)
BandwidthSummarySatellite(ctx context.Context, in *BandwidthSummarySatelliteRequest) (*BandwidthSummarySatelliteResponse, error)
BandwidthSummary(ctx context.Context, in *BandwidthSummaryRequest) (*BandwidthSummaryResponse, error)
EgressSummarySatellite(ctx context.Context, in *EgressSummarySatelliteRequest) (*EgressSummarySatelliteResponse, error)
EgressSummary(ctx context.Context, in *EgressSummaryRequest) (*EgressSummaryResponse, error)
IngressSummarySatellite(ctx context.Context, in *IngressSummarySatelliteRequest) (*IngressSummarySatelliteResponse, error)
IngressSummary(ctx context.Context, in *IngressSummaryRequest) (*IngressSummaryResponse, error)
DailySatellite(ctx context.Context, in *DailySatelliteRequest) (*DailySatelliteResponse, error)
Daily(ctx context.Context, in *DailyRequest) (*DailyResponse, error)
}
type drpcBandwidthClient struct {
@ -219,8 +227,88 @@ func (c *drpcBandwidthClient) MonthSummary(ctx context.Context, in *BandwidthMon
return out, nil
}
func (c *drpcBandwidthClient) BandwidthSummarySatellite(ctx context.Context, in *BandwidthSummarySatelliteRequest) (*BandwidthSummarySatelliteResponse, error) {
out := new(BandwidthSummarySatelliteResponse)
err := c.cc.Invoke(ctx, "/multinode.Bandwidth/BandwidthSummarySatellite", drpcEncoding_File_multinode_proto{}, in, out)
if err != nil {
return nil, err
}
return out, nil
}
func (c *drpcBandwidthClient) BandwidthSummary(ctx context.Context, in *BandwidthSummaryRequest) (*BandwidthSummaryResponse, error) {
out := new(BandwidthSummaryResponse)
err := c.cc.Invoke(ctx, "/multinode.Bandwidth/BandwidthSummary", drpcEncoding_File_multinode_proto{}, in, out)
if err != nil {
return nil, err
}
return out, nil
}
func (c *drpcBandwidthClient) EgressSummarySatellite(ctx context.Context, in *EgressSummarySatelliteRequest) (*EgressSummarySatelliteResponse, error) {
out := new(EgressSummarySatelliteResponse)
err := c.cc.Invoke(ctx, "/multinode.Bandwidth/EgressSummarySatellite", drpcEncoding_File_multinode_proto{}, in, out)
if err != nil {
return nil, err
}
return out, nil
}
func (c *drpcBandwidthClient) EgressSummary(ctx context.Context, in *EgressSummaryRequest) (*EgressSummaryResponse, error) {
out := new(EgressSummaryResponse)
err := c.cc.Invoke(ctx, "/multinode.Bandwidth/EgressSummary", drpcEncoding_File_multinode_proto{}, in, out)
if err != nil {
return nil, err
}
return out, nil
}
func (c *drpcBandwidthClient) IngressSummarySatellite(ctx context.Context, in *IngressSummarySatelliteRequest) (*IngressSummarySatelliteResponse, error) {
out := new(IngressSummarySatelliteResponse)
err := c.cc.Invoke(ctx, "/multinode.Bandwidth/IngressSummarySatellite", drpcEncoding_File_multinode_proto{}, in, out)
if err != nil {
return nil, err
}
return out, nil
}
func (c *drpcBandwidthClient) IngressSummary(ctx context.Context, in *IngressSummaryRequest) (*IngressSummaryResponse, error) {
out := new(IngressSummaryResponse)
err := c.cc.Invoke(ctx, "/multinode.Bandwidth/IngressSummary", drpcEncoding_File_multinode_proto{}, in, out)
if err != nil {
return nil, err
}
return out, nil
}
func (c *drpcBandwidthClient) DailySatellite(ctx context.Context, in *DailySatelliteRequest) (*DailySatelliteResponse, error) {
out := new(DailySatelliteResponse)
err := c.cc.Invoke(ctx, "/multinode.Bandwidth/DailySatellite", drpcEncoding_File_multinode_proto{}, in, out)
if err != nil {
return nil, err
}
return out, nil
}
func (c *drpcBandwidthClient) Daily(ctx context.Context, in *DailyRequest) (*DailyResponse, error) {
out := new(DailyResponse)
err := c.cc.Invoke(ctx, "/multinode.Bandwidth/Daily", drpcEncoding_File_multinode_proto{}, in, out)
if err != nil {
return nil, err
}
return out, nil
}
type DRPCBandwidthServer interface {
MonthSummary(context.Context, *BandwidthMonthSummaryRequest) (*BandwidthMonthSummaryResponse, error)
BandwidthSummarySatellite(context.Context, *BandwidthSummarySatelliteRequest) (*BandwidthSummarySatelliteResponse, error)
BandwidthSummary(context.Context, *BandwidthSummaryRequest) (*BandwidthSummaryResponse, error)
EgressSummarySatellite(context.Context, *EgressSummarySatelliteRequest) (*EgressSummarySatelliteResponse, error)
EgressSummary(context.Context, *EgressSummaryRequest) (*EgressSummaryResponse, error)
IngressSummarySatellite(context.Context, *IngressSummarySatelliteRequest) (*IngressSummarySatelliteResponse, error)
IngressSummary(context.Context, *IngressSummaryRequest) (*IngressSummaryResponse, error)
DailySatellite(context.Context, *DailySatelliteRequest) (*DailySatelliteResponse, error)
Daily(context.Context, *DailyRequest) (*DailyResponse, error)
}
type DRPCBandwidthUnimplementedServer struct{}
@ -229,9 +317,41 @@ func (s *DRPCBandwidthUnimplementedServer) MonthSummary(context.Context, *Bandwi
return nil, drpcerr.WithCode(errors.New("Unimplemented"), 12)
}
func (s *DRPCBandwidthUnimplementedServer) BandwidthSummarySatellite(context.Context, *BandwidthSummarySatelliteRequest) (*BandwidthSummarySatelliteResponse, error) {
return nil, drpcerr.WithCode(errors.New("Unimplemented"), 12)
}
func (s *DRPCBandwidthUnimplementedServer) BandwidthSummary(context.Context, *BandwidthSummaryRequest) (*BandwidthSummaryResponse, error) {
return nil, drpcerr.WithCode(errors.New("Unimplemented"), 12)
}
func (s *DRPCBandwidthUnimplementedServer) EgressSummarySatellite(context.Context, *EgressSummarySatelliteRequest) (*EgressSummarySatelliteResponse, error) {
return nil, drpcerr.WithCode(errors.New("Unimplemented"), 12)
}
func (s *DRPCBandwidthUnimplementedServer) EgressSummary(context.Context, *EgressSummaryRequest) (*EgressSummaryResponse, error) {
return nil, drpcerr.WithCode(errors.New("Unimplemented"), 12)
}
func (s *DRPCBandwidthUnimplementedServer) IngressSummarySatellite(context.Context, *IngressSummarySatelliteRequest) (*IngressSummarySatelliteResponse, error) {
return nil, drpcerr.WithCode(errors.New("Unimplemented"), 12)
}
func (s *DRPCBandwidthUnimplementedServer) IngressSummary(context.Context, *IngressSummaryRequest) (*IngressSummaryResponse, error) {
return nil, drpcerr.WithCode(errors.New("Unimplemented"), 12)
}
func (s *DRPCBandwidthUnimplementedServer) DailySatellite(context.Context, *DailySatelliteRequest) (*DailySatelliteResponse, error) {
return nil, drpcerr.WithCode(errors.New("Unimplemented"), 12)
}
func (s *DRPCBandwidthUnimplementedServer) Daily(context.Context, *DailyRequest) (*DailyResponse, error) {
return nil, drpcerr.WithCode(errors.New("Unimplemented"), 12)
}
type DRPCBandwidthDescription struct{}
func (DRPCBandwidthDescription) NumMethods() int { return 1 }
func (DRPCBandwidthDescription) NumMethods() int { return 9 }
func (DRPCBandwidthDescription) Method(n int) (string, drpc.Encoding, drpc.Receiver, interface{}, bool) {
switch n {
@ -244,6 +364,78 @@ func (DRPCBandwidthDescription) Method(n int) (string, drpc.Encoding, drpc.Recei
in1.(*BandwidthMonthSummaryRequest),
)
}, DRPCBandwidthServer.MonthSummary, true
case 1:
return "/multinode.Bandwidth/BandwidthSummarySatellite", drpcEncoding_File_multinode_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return srv.(DRPCBandwidthServer).
BandwidthSummarySatellite(
ctx,
in1.(*BandwidthSummarySatelliteRequest),
)
}, DRPCBandwidthServer.BandwidthSummarySatellite, true
case 2:
return "/multinode.Bandwidth/BandwidthSummary", drpcEncoding_File_multinode_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return srv.(DRPCBandwidthServer).
BandwidthSummary(
ctx,
in1.(*BandwidthSummaryRequest),
)
}, DRPCBandwidthServer.BandwidthSummary, true
case 3:
return "/multinode.Bandwidth/EgressSummarySatellite", drpcEncoding_File_multinode_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return srv.(DRPCBandwidthServer).
EgressSummarySatellite(
ctx,
in1.(*EgressSummarySatelliteRequest),
)
}, DRPCBandwidthServer.EgressSummarySatellite, true
case 4:
return "/multinode.Bandwidth/EgressSummary", drpcEncoding_File_multinode_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return srv.(DRPCBandwidthServer).
EgressSummary(
ctx,
in1.(*EgressSummaryRequest),
)
}, DRPCBandwidthServer.EgressSummary, true
case 5:
return "/multinode.Bandwidth/IngressSummarySatellite", drpcEncoding_File_multinode_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return srv.(DRPCBandwidthServer).
IngressSummarySatellite(
ctx,
in1.(*IngressSummarySatelliteRequest),
)
}, DRPCBandwidthServer.IngressSummarySatellite, true
case 6:
return "/multinode.Bandwidth/IngressSummary", drpcEncoding_File_multinode_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return srv.(DRPCBandwidthServer).
IngressSummary(
ctx,
in1.(*IngressSummaryRequest),
)
}, DRPCBandwidthServer.IngressSummary, true
case 7:
return "/multinode.Bandwidth/DailySatellite", drpcEncoding_File_multinode_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return srv.(DRPCBandwidthServer).
DailySatellite(
ctx,
in1.(*DailySatelliteRequest),
)
}, DRPCBandwidthServer.DailySatellite, true
case 8:
return "/multinode.Bandwidth/Daily", drpcEncoding_File_multinode_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return srv.(DRPCBandwidthServer).
Daily(
ctx,
in1.(*DailyRequest),
)
}, DRPCBandwidthServer.Daily, true
default:
return "", nil, nil, nil, false
}
@ -269,6 +461,134 @@ func (x *drpcBandwidth_MonthSummaryStream) SendAndClose(m *BandwidthMonthSummary
return x.CloseSend()
}
type DRPCBandwidth_BandwidthSummarySatelliteStream interface {
drpc.Stream
SendAndClose(*BandwidthSummarySatelliteResponse) error
}
type drpcBandwidth_BandwidthSummarySatelliteStream struct {
drpc.Stream
}
func (x *drpcBandwidth_BandwidthSummarySatelliteStream) SendAndClose(m *BandwidthSummarySatelliteResponse) error {
if err := x.MsgSend(m, drpcEncoding_File_multinode_proto{}); err != nil {
return err
}
return x.CloseSend()
}
type DRPCBandwidth_BandwidthSummaryStream interface {
drpc.Stream
SendAndClose(*BandwidthSummaryResponse) error
}
type drpcBandwidth_BandwidthSummaryStream struct {
drpc.Stream
}
func (x *drpcBandwidth_BandwidthSummaryStream) SendAndClose(m *BandwidthSummaryResponse) error {
if err := x.MsgSend(m, drpcEncoding_File_multinode_proto{}); err != nil {
return err
}
return x.CloseSend()
}
type DRPCBandwidth_EgressSummarySatelliteStream interface {
drpc.Stream
SendAndClose(*EgressSummarySatelliteResponse) error
}
type drpcBandwidth_EgressSummarySatelliteStream struct {
drpc.Stream
}
func (x *drpcBandwidth_EgressSummarySatelliteStream) SendAndClose(m *EgressSummarySatelliteResponse) error {
if err := x.MsgSend(m, drpcEncoding_File_multinode_proto{}); err != nil {
return err
}
return x.CloseSend()
}
type DRPCBandwidth_EgressSummaryStream interface {
drpc.Stream
SendAndClose(*EgressSummaryResponse) error
}
type drpcBandwidth_EgressSummaryStream struct {
drpc.Stream
}
func (x *drpcBandwidth_EgressSummaryStream) SendAndClose(m *EgressSummaryResponse) error {
if err := x.MsgSend(m, drpcEncoding_File_multinode_proto{}); err != nil {
return err
}
return x.CloseSend()
}
type DRPCBandwidth_IngressSummarySatelliteStream interface {
drpc.Stream
SendAndClose(*IngressSummarySatelliteResponse) error
}
type drpcBandwidth_IngressSummarySatelliteStream struct {
drpc.Stream
}
func (x *drpcBandwidth_IngressSummarySatelliteStream) SendAndClose(m *IngressSummarySatelliteResponse) error {
if err := x.MsgSend(m, drpcEncoding_File_multinode_proto{}); err != nil {
return err
}
return x.CloseSend()
}
type DRPCBandwidth_IngressSummaryStream interface {
drpc.Stream
SendAndClose(*IngressSummaryResponse) error
}
type drpcBandwidth_IngressSummaryStream struct {
drpc.Stream
}
func (x *drpcBandwidth_IngressSummaryStream) SendAndClose(m *IngressSummaryResponse) error {
if err := x.MsgSend(m, drpcEncoding_File_multinode_proto{}); err != nil {
return err
}
return x.CloseSend()
}
type DRPCBandwidth_DailySatelliteStream interface {
drpc.Stream
SendAndClose(*DailySatelliteResponse) error
}
type drpcBandwidth_DailySatelliteStream struct {
drpc.Stream
}
func (x *drpcBandwidth_DailySatelliteStream) SendAndClose(m *DailySatelliteResponse) error {
if err := x.MsgSend(m, drpcEncoding_File_multinode_proto{}); err != nil {
return err
}
return x.CloseSend()
}
type DRPCBandwidth_DailyStream interface {
drpc.Stream
SendAndClose(*DailyResponse) error
}
type drpcBandwidth_DailyStream struct {
drpc.Stream
}
func (x *drpcBandwidth_DailyStream) SendAndClose(m *DailyResponse) error {
if err := x.MsgSend(m, drpcEncoding_File_multinode_proto{}); err != nil {
return err
}
return x.CloseSend()
}
type DRPCNodeClient interface {
DRPCConn() drpc.Conn

179
storagenode/multinode/bandwidth.go
View File

@ -10,6 +10,7 @@ import (
"go.uber.org/zap"
"storj.io/common/rpc/rpcstatus"
"storj.io/storj/private/date"
"storj.io/storj/private/multinodepb"
"storj.io/storj/storagenode/apikeys"
"storj.io/storj/storagenode/bandwidth"
@ -54,3 +55,181 @@ func (bandwidth *BandwidthEndpoint) MonthSummary(ctx context.Context, req *multi
Used: used,
}, nil
}
// BandwidthSummarySatellite returns bandwidth summary for specific satellite.
func (bandwidth *BandwidthEndpoint) BandwidthSummarySatellite(ctx context.Context, req *multinodepb.BandwidthSummarySatelliteRequest) (_ *multinodepb.BandwidthSummarySatelliteResponse, err error) {
defer mon.Task()(&ctx)(&err)
if err = authenticate(ctx, bandwidth.apiKeys, req.GetHeader()); err != nil {
return nil, rpcstatus.Wrap(rpcstatus.Unauthenticated, err)
}
from, to := date.MonthBoundary(time.Now().UTC())
bandwidthSummary, err := bandwidth.db.SatelliteSummary(ctx, req.SatelliteId, from, to)
if err != nil {
bandwidth.log.Error("bandwidth internal error", zap.Error(err))
return nil, rpcstatus.Wrap(rpcstatus.Internal, err)
}
return &multinodepb.BandwidthSummarySatelliteResponse{Summary: bandwidthSummary.Total()}, nil
}
// BandwidthSummary returns bandwidth summary.
func (bandwidth *BandwidthEndpoint) BandwidthSummary(ctx context.Context, req *multinodepb.BandwidthSummaryRequest) (_ *multinodepb.BandwidthSummaryResponse, err error) {
defer mon.Task()(&ctx)(&err)
if err = authenticate(ctx, bandwidth.apiKeys, req.GetHeader()); err != nil {
return nil, rpcstatus.Wrap(rpcstatus.Unauthenticated, err)
}
from, to := date.MonthBoundary(time.Now().UTC())
bandwidthSummary, err := bandwidth.db.Summary(ctx, from, to)
if err != nil {
bandwidth.log.Error("bandwidth internal error", zap.Error(err))
return nil, rpcstatus.Wrap(rpcstatus.Internal, err)
}
return &multinodepb.BandwidthSummaryResponse{Summary: bandwidthSummary.Total()}, nil
}
// EgressSummarySatellite returns egress summary for specific satellite.
func (bandwidth *BandwidthEndpoint) EgressSummarySatellite(ctx context.Context, req *multinodepb.EgressSummarySatelliteRequest) (_ *multinodepb.EgressSummarySatelliteResponse, err error) {
defer mon.Task()(&ctx)(&err)
if err = authenticate(ctx, bandwidth.apiKeys, req.GetHeader()); err != nil {
return nil, rpcstatus.Wrap(rpcstatus.Unauthenticated, err)
}
from, to := date.MonthBoundary(time.Now().UTC())
egressSummary, err := bandwidth.db.SatelliteEgressSummary(ctx, req.SatelliteId, from, to)
if err != nil {
bandwidth.log.Error("bandwidth internal error", zap.Error(err))
return nil, rpcstatus.Wrap(rpcstatus.Internal, err)
}
return &multinodepb.EgressSummarySatelliteResponse{Summary: egressSummary.Total()}, nil
}
// EgressSummary returns egress summary.
func (bandwidth *BandwidthEndpoint) EgressSummary(ctx context.Context, req *multinodepb.EgressSummaryRequest) (_ *multinodepb.EgressSummaryResponse, err error) {
defer mon.Task()(&ctx)(&err)
if err = authenticate(ctx, bandwidth.apiKeys, req.GetHeader()); err != nil {
return nil, rpcstatus.Wrap(rpcstatus.Unauthenticated, err)
}
from, to := date.MonthBoundary(time.Now().UTC())
egressSummary, err := bandwidth.db.EgressSummary(ctx, from, to)
if err != nil {
bandwidth.log.Error("bandwidth internal error", zap.Error(err))
return nil, rpcstatus.Wrap(rpcstatus.Internal, err)
}
return &multinodepb.EgressSummaryResponse{Summary: egressSummary.Total()}, nil
}
// IngressSummarySatellite returns ingress summary for specific satellite.
func (bandwidth *BandwidthEndpoint) IngressSummarySatellite(ctx context.Context, req *multinodepb.IngressSummarySatelliteRequest) (_ *multinodepb.IngressSummarySatelliteResponse, err error) {
defer mon.Task()(&ctx)(&err)
if err = authenticate(ctx, bandwidth.apiKeys, req.GetHeader()); err != nil {
return nil, rpcstatus.Wrap(rpcstatus.Unauthenticated, err)
}
from, to := date.MonthBoundary(time.Now().UTC())
ingressSummary, err := bandwidth.db.SatelliteIngressSummary(ctx, req.SatelliteId, from, to)
if err != nil {
bandwidth.log.Error("bandwidth internal error", zap.Error(err))
return nil, rpcstatus.Wrap(rpcstatus.Internal, err)
}
return &multinodepb.IngressSummarySatelliteResponse{Summary: ingressSummary.Total()}, nil
}
// IngressSummary returns ingress summary.
func (bandwidth *BandwidthEndpoint) IngressSummary(ctx context.Context, req *multinodepb.IngressSummaryRequest) (_ *multinodepb.IngressSummaryResponse, err error) {
defer mon.Task()(&ctx)(&err)
if err = authenticate(ctx, bandwidth.apiKeys, req.GetHeader()); err != nil {
return nil, rpcstatus.Wrap(rpcstatus.Unauthenticated, err)
}
from, to := date.MonthBoundary(time.Now().UTC())
ingressSummary, err := bandwidth.db.IngressSummary(ctx, from, to)
if err != nil {
bandwidth.log.Error("bandwidth internal error", zap.Error(err))
return nil, rpcstatus.Wrap(rpcstatus.Internal, err)
}
return &multinodepb.IngressSummaryResponse{Summary: ingressSummary.Total()}, nil
}
// DailySatellite returns bandwidth summary split by days current month for specific satellite.
func (bandwidth *BandwidthEndpoint) DailySatellite(ctx context.Context, req *multinodepb.DailySatelliteRequest) (_ *multinodepb.DailySatelliteResponse, err error) {
defer mon.Task()(&ctx)(&err)
if err = authenticate(ctx, bandwidth.apiKeys, req.GetHeader()); err != nil {
return nil, rpcstatus.Wrap(rpcstatus.Unauthenticated, err)
}
from, to := date.MonthBoundary(time.Now().UTC())
bandwidthDaily, err := bandwidth.db.GetDailySatelliteRollups(ctx, req.SatelliteId, from, to)
if err != nil {
bandwidth.log.Error("bandwidth internal error", zap.Error(err))
return nil, rpcstatus.Wrap(rpcstatus.Internal, err)
}
var resp []*multinodepb.UsageRollup
for _, bd := range bandwidthDaily {
resp = append(resp, &multinodepb.UsageRollup{
Egress: &multinodepb.Egress{
Repair: bd.Egress.Repair,
Audit: bd.Egress.Audit,
Usage: bd.Egress.Usage,
},
Ingress: &multinodepb.Ingress{
Repaid: bd.Ingress.Repair,
Usage: bd.Ingress.Usage,
},
Delete: bd.Delete,
IntervalStart: bd.IntervalStart,
})
}
return &multinodepb.DailySatelliteResponse{UsageRollup: resp}, nil
}
// Daily returns bandwidth summary split by days current month.
func (bandwidth *BandwidthEndpoint) Daily(ctx context.Context, req *multinodepb.DailyRequest) (_ *multinodepb.DailyResponse, err error) {
defer mon.Task()(&ctx)(&err)
if err = authenticate(ctx, bandwidth.apiKeys, req.GetHeader()); err != nil {
return nil, rpcstatus.Wrap(rpcstatus.Unauthenticated, err)
}
from, to := date.MonthBoundary(time.Now().UTC())
bandwidthDaily, err := bandwidth.db.GetDailyRollups(ctx, from, to)
if err != nil {
bandwidth.log.Error("bandwidth internal error", zap.Error(err))
return nil, rpcstatus.Wrap(rpcstatus.Internal, err)
}
var resp []*multinodepb.UsageRollup
for _, bd := range bandwidthDaily {
resp = append(resp, &multinodepb.UsageRollup{
Egress: &multinodepb.Egress{
Repair: bd.Egress.Repair,
Audit: bd.Egress.Audit,
Usage: bd.Egress.Usage,
},
Ingress: &multinodepb.Ingress{
Repaid: bd.Ingress.Repair,
Usage: bd.Ingress.Usage,
},
Delete: bd.Delete,
IntervalStart: bd.IntervalStart,
})
}
return &multinodepb.DailyResponse{UsageRollup: resp}, nil
}