storj/satellite/orders/endpoint.go
Michal Niewrzal 15508d270c satellite/orders: don't store non user bandwidth actions for bucket
For bucket_bandwidth_rollups we are trying to insert lots of entries
with empty bucket name and project id. Those are inserts from orders
created by repair, audit and GE. High load on the same primary key
(the same range) is causing many retries and that's affect all inserts
as we are putting 1000 entries into DB a once.

This change solves this problem by not storing into
bucket_bandwidth_rollups other actions then GET and PUT. Those actions
are only important from bucket bandwidth usage perspective because
those are actions performed by users. Other actions (repair, audit or
GE) are also stored in storagenode_bandwdith_rollups so we will still
have access to them e.g. for statistic purposes.

https://github.com/storj/storj/issues/5332

Change-Id: Ibb5bf0a4c869b0439dc65da1c9342a38ca2890ba
2023-02-01 15:38:48 +00:00

454 lines
16 KiB
Go

// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package orders
import (
"bytes"
"context"
"errors"
"io"
"sort"
"time"
"github.com/spacemonkeygo/monkit/v3"
"github.com/zeebo/errs"
"go.uber.org/zap"
"storj.io/common/identity"
"storj.io/common/pb"
"storj.io/common/rpc/rpcstatus"
"storj.io/common/signing"
"storj.io/common/storj"
"storj.io/common/uuid"
"storj.io/storj/private/date"
"storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/nodeapiversion"
)
// DB implements saving order after receiving from storage node.
//
// architecture: Database
type DB interface {
// UpdateBucketBandwidthAllocation updates 'allocated' bandwidth for given bucket
UpdateBucketBandwidthAllocation(ctx context.Context, projectID uuid.UUID, bucketName []byte, action pb.PieceAction, amount int64, intervalStart time.Time) error
// UpdateBucketBandwidthSettle updates 'settled' bandwidth for given bucket
UpdateBucketBandwidthSettle(ctx context.Context, projectID uuid.UUID, bucketName []byte, action pb.PieceAction, settledAmount, deadAmount int64, intervalStart time.Time) error
// UpdateBucketBandwidthInline updates 'inline' bandwidth for given bucket
UpdateBucketBandwidthInline(ctx context.Context, projectID uuid.UUID, bucketName []byte, action pb.PieceAction, amount int64, intervalStart time.Time) error
// UpdateBandwidthBatch updates bucket and project bandwidth rollups in the database
UpdateBandwidthBatch(ctx context.Context, rollups []BucketBandwidthRollup) error
// UpdateStoragenodeBandwidthSettle updates 'settled' bandwidth for given storage node
UpdateStoragenodeBandwidthSettle(ctx context.Context, storageNode storj.NodeID, action pb.PieceAction, amount int64, intervalStart time.Time) error
// UpdateStoragenodeBandwidthSettleWithWindow updates 'settled' bandwidth for given storage node
UpdateStoragenodeBandwidthSettleWithWindow(ctx context.Context, storageNodeID storj.NodeID, actionAmounts map[int32]int64, window time.Time) (status pb.SettlementWithWindowResponse_Status, alreadyProcessed bool, err error)
// GetBucketBandwidth gets total bucket bandwidth from period of time
GetBucketBandwidth(ctx context.Context, projectID uuid.UUID, bucketName []byte, from, to time.Time) (int64, error)
// GetStorageNodeBandwidth gets total storage node bandwidth from period of time
GetStorageNodeBandwidth(ctx context.Context, nodeID storj.NodeID, from, to time.Time) (int64, error)
}
type noopDB struct {
}
func (noopDB) UpdateBucketBandwidthAllocation(ctx context.Context, projectID uuid.UUID, bucketName []byte, action pb.PieceAction, amount int64, intervalStart time.Time) error {
return nil
}
func (noopDB) UpdateBucketBandwidthSettle(ctx context.Context, projectID uuid.UUID, bucketName []byte, action pb.PieceAction, settledAmount, deadAmount int64, intervalStart time.Time) error {
return nil
}
func (noopDB) UpdateBucketBandwidthInline(ctx context.Context, projectID uuid.UUID, bucketName []byte, action pb.PieceAction, amount int64, intervalStart time.Time) error {
return nil
}
func (noopDB) UpdateBandwidthBatch(ctx context.Context, rollups []BucketBandwidthRollup) error {
return nil
}
func (noopDB) UpdateStoragenodeBandwidthSettle(ctx context.Context, storageNode storj.NodeID, action pb.PieceAction, amount int64, intervalStart time.Time) error {
return nil
}
func (noopDB) UpdateStoragenodeBandwidthSettleWithWindow(ctx context.Context, storageNodeID storj.NodeID, actionAmounts map[int32]int64, window time.Time) (status pb.SettlementWithWindowResponse_Status, alreadyProcessed bool, err error) {
return pb.SettlementWithWindowResponse_ACCEPTED, false, nil
}
func (noopDB) GetBucketBandwidth(ctx context.Context, projectID uuid.UUID, bucketName []byte, from, to time.Time) (int64, error) {
return 0, nil
}
func (noopDB) GetStorageNodeBandwidth(ctx context.Context, nodeID storj.NodeID, from, to time.Time) (int64, error) {
return 0, nil
}
// NewNoopDB creates noop orders DB.
func NewNoopDB() DB {
return &noopDB{}
}
// SerialDeleteOptions are option when deleting from serial tables.
type SerialDeleteOptions struct {
BatchSize int
}
// ConsumedSerial is a serial that has been consumed and its bandwidth recorded.
type ConsumedSerial struct {
NodeID storj.NodeID
SerialNumber storj.SerialNumber
ExpiresAt time.Time
}
// PendingSerial is a serial number reported by a storagenode waiting to be
// settled.
type PendingSerial struct {
NodeID storj.NodeID
BucketID []byte
Action uint
SerialNumber storj.SerialNumber
ExpiresAt time.Time
Settled uint64
}
var (
// Error the default orders errs class.
Error = errs.Class("orders")
// ErrUsingSerialNumber error class for serial number.
ErrUsingSerialNumber = errs.Class("serial number")
mon = monkit.Package()
)
// BucketBandwidthRollup contains all the info needed for a bucket bandwidth rollup.
type BucketBandwidthRollup struct {
ProjectID uuid.UUID
BucketName string
Action pb.PieceAction
IntervalStart time.Time
Inline int64
Allocated int64
Settled int64
Dead int64
}
// StoragenodeBandwidthRollup contains all the info needed for a storagenode bandwidth rollup.
type StoragenodeBandwidthRollup struct {
NodeID storj.NodeID
Action pb.PieceAction
Allocated int64
Settled int64
}
// SortStoragenodeBandwidthRollups sorts the rollups.
func SortStoragenodeBandwidthRollups(rollups []StoragenodeBandwidthRollup) {
sort.SliceStable(rollups, func(i, j int) bool {
nodeCompare := bytes.Compare(rollups[i].NodeID.Bytes(), rollups[j].NodeID.Bytes())
switch {
case nodeCompare == -1:
return true
case nodeCompare == 1:
return false
case rollups[i].Action < rollups[j].Action:
return true
case rollups[i].Action > rollups[j].Action:
return false
default:
return false
}
})
}
// Endpoint for orders receiving.
//
// architecture: Endpoint
type Endpoint struct {
pb.DRPCOrdersUnimplementedServer
log *zap.Logger
satelliteSignee signing.Signee
DB DB
nodeAPIVersionDB nodeapiversion.DB
ordersSemaphore chan struct{}
ordersService *Service
}
// NewEndpoint new orders receiving endpoint.
//
// ordersSemaphoreSize controls the number of concurrent clients allowed to submit orders at once.
// A value of zero means unlimited.
func NewEndpoint(log *zap.Logger, satelliteSignee signing.Signee, db DB, nodeAPIVersionDB nodeapiversion.DB,
ordersSemaphoreSize int, ordersService *Service) *Endpoint {
var ordersSemaphore chan struct{}
if ordersSemaphoreSize > 0 {
ordersSemaphore = make(chan struct{}, ordersSemaphoreSize)
}
return &Endpoint{
log: log,
satelliteSignee: satelliteSignee,
DB: db,
nodeAPIVersionDB: nodeAPIVersionDB,
ordersSemaphore: ordersSemaphore,
ordersService: ordersService,
}
}
type bucketIDAction struct {
bucketname string
projectID uuid.UUID
action pb.PieceAction
}
// SettlementWithWindow processes all orders that were created in a 1 hour window.
// Only one window is processed at a time.
// Batches are atomic, all orders are settled successfully or they all fail.
func (endpoint *Endpoint) SettlementWithWindow(stream pb.DRPCOrders_SettlementWithWindowStream) (err error) {
return endpoint.SettlementWithWindowFinal(stream)
}
func trackFinalStatus(status pb.SettlementWithWindowResponse_Status) {
switch status {
case pb.SettlementWithWindowResponse_ACCEPTED:
mon.Event("settlement_response_accepted")
case pb.SettlementWithWindowResponse_REJECTED:
mon.Event("settlement_response_rejected")
default:
mon.Event("settlement_response_unknown")
}
}
// SettlementWithWindowFinal processes all orders that were created in a 1 hour window.
// Only one window is processed at a time.
// Batches are atomic, all orders are settled successfully or they all fail.
func (endpoint *Endpoint) SettlementWithWindowFinal(stream pb.DRPCOrders_SettlementWithWindowStream) (err error) {
ctx := stream.Context()
defer mon.Task()(&ctx)(&err)
var alreadyProcessed bool
var status pb.SettlementWithWindowResponse_Status
defer trackFinalStatus(status)
peer, err := identity.PeerIdentityFromContext(ctx)
if err != nil {
endpoint.log.Debug("err peer identity from context", zap.Error(err))
return rpcstatus.Error(rpcstatus.Unauthenticated, err.Error())
}
versionAtLeast, err := endpoint.nodeAPIVersionDB.VersionAtLeast(ctx, peer.ID, nodeapiversion.HasWindowedOrders)
if err != nil {
endpoint.log.Info("could not query if node version was new enough", zap.Error(err))
versionAtLeast = false
}
if !versionAtLeast {
err = endpoint.nodeAPIVersionDB.UpdateVersionAtLeast(ctx, peer.ID, nodeapiversion.HasWindowedOrders)
if err != nil {
return rpcstatus.Wrap(rpcstatus.Internal, err)
}
}
log := endpoint.log.Named(peer.ID.String())
log.Debug("SettlementWithWindow")
type bandwidthAmount struct {
Settled int64
Dead int64
}
storagenodeSettled := map[int32]int64{}
bucketSettled := map[bucketIDAction]bandwidthAmount{}
seenSerials := map[storj.SerialNumber]struct{}{}
var window int64
var request *pb.SettlementRequest
var receivedCount int
for {
request, err = stream.Recv()
if err != nil {
if errors.Is(err, io.EOF) {
break
}
log.Debug("err streaming order request", zap.Error(err))
return rpcstatus.Error(rpcstatus.Unknown, err.Error())
}
receivedCount++
orderLimit := request.Limit
if orderLimit == nil {
log.Debug("request.OrderLimit is nil")
continue
}
if window == 0 {
window = date.TruncateToHourInNano(orderLimit.OrderCreation)
}
order := request.Order
if order == nil {
log.Debug("request.Order is nil")
continue
}
serialNum := order.SerialNumber
// don't process orders that aren't valid
if !endpoint.isValid(ctx, log, order, orderLimit, peer.ID, window) {
continue
}
// don't process orders with serial numbers we've already seen
if _, ok := seenSerials[serialNum]; ok {
log.Debug("seen serial", zap.String("serial number", serialNum.String()))
continue
}
seenSerials[serialNum] = struct{}{}
storagenodeSettled[int32(orderLimit.Action)] += order.Amount
// user can do only two actions which are important for bucket bandwidth usage
userAction := orderLimit.Action == pb.PieceAction_PUT || orderLimit.Action == pb.PieceAction_GET
// don't store anything else than user actions in bucket_bandwidth_rollups table. amounts for other
// actions will be stored in storagenode_bandwidth_rollups.
if !userAction {
continue
}
metadata, err := endpoint.ordersService.DecryptOrderMetadata(ctx, orderLimit)
if err != nil {
log.Debug("decrypt order metadata err:", zap.Error(err))
mon.Event("bucketinfo_from_orders_metadata_error_1")
continue
}
var bucketInfo metabase.BucketLocation
switch {
case len(metadata.CompactProjectBucketPrefix) > 0:
bucketInfo, err = metabase.ParseCompactBucketPrefix(metadata.GetCompactProjectBucketPrefix())
if err != nil {
log.Debug("decrypt order: ParseCompactBucketPrefix", zap.Error(err))
mon.Event("bucketinfo_from_orders_metadata_error_compact")
continue
}
case len(metadata.ProjectBucketPrefix) > 0:
bucketInfo, err = metabase.ParseBucketPrefix(metabase.BucketPrefix(metadata.GetProjectBucketPrefix()))
if err != nil {
log.Debug("decrypt order: ParseBucketPrefix", zap.Error(err))
mon.Event("bucketinfo_from_orders_metadata_error_uncompact")
continue
}
default:
log.Debug("decrypt order: project bucket prefix missing", zap.Error(err))
mon.Event("bucketinfo_from_orders_metadata_error_default")
continue
}
// log error only for orders created by users, for satellite actions order limits are created
// without bucket name and project ID because segments loop doesn't have access to it
if bucketInfo.BucketName == "" || bucketInfo.ProjectID.IsZero() {
log.Warn("decrypt order: bucketName or projectID not set",
zap.String("bucketName", bucketInfo.BucketName),
zap.String("projectID", bucketInfo.ProjectID.String()),
)
mon.Event("bucketinfo_from_orders_metadata_error_3")
continue
}
currentBucketIDAction := bucketIDAction{
bucketname: bucketInfo.BucketName,
projectID: bucketInfo.ProjectID,
action: orderLimit.Action,
}
bucketSettled[currentBucketIDAction] = bandwidthAmount{
Settled: bucketSettled[currentBucketIDAction].Settled + order.Amount,
Dead: bucketSettled[currentBucketIDAction].Dead + orderLimit.Limit - order.Amount,
// we are not collecting Allocated bandwidth as it won't be stored with UpdateBucketBandwidthSettle
}
}
if len(storagenodeSettled) == 0 {
log.Debug("no orders were successfully processed", zap.Int("received count", receivedCount))
status = pb.SettlementWithWindowResponse_REJECTED
return stream.SendAndClose(&pb.SettlementWithWindowResponse{
Status: status,
ActionSettled: storagenodeSettled,
})
}
status, alreadyProcessed, err = endpoint.DB.UpdateStoragenodeBandwidthSettleWithWindow(
ctx, peer.ID, storagenodeSettled, time.Unix(0, window),
)
if err != nil {
log.Debug("err updating storagenode bandwidth settle", zap.Error(err))
return err
}
log.Debug("orders processed",
zap.Int("total orders received", receivedCount),
zap.Time("window", time.Unix(0, window)),
zap.String("status", status.String()),
)
if status == pb.SettlementWithWindowResponse_ACCEPTED && !alreadyProcessed {
for bucketIDAction, bwAmount := range bucketSettled {
err = endpoint.DB.UpdateBucketBandwidthSettle(ctx,
bucketIDAction.projectID, []byte(bucketIDAction.bucketname), bucketIDAction.action, bwAmount.Settled, bwAmount.Dead, time.Unix(0, window),
)
if err != nil {
log.Info("err updating bucket bandwidth settle", zap.Error(err))
}
}
} else {
mon.Event("orders_already_processed")
}
if status == pb.SettlementWithWindowResponse_REJECTED {
storagenodeSettled = map[int32]int64{}
}
return stream.SendAndClose(&pb.SettlementWithWindowResponse{
Status: status,
ActionSettled: storagenodeSettled,
})
}
func (endpoint *Endpoint) isValid(ctx context.Context, log *zap.Logger, order *pb.Order,
orderLimit *pb.OrderLimit, peerID storj.NodeID, window int64) bool {
if orderLimit.StorageNodeId != peerID {
log.Debug("storage node id mismatch")
mon.Event("order_not_valid_storagenodeid")
return false
}
// check expiration first before the signatures so that we can throw out the large amount
// of expired orders being sent to us before doing expensive signature verification.
if orderLimit.OrderExpiration.Before(time.Now().UTC()) {
log.Debug("invalid settlement: order limit expired")
mon.Event("order_not_valid_expired")
return false
}
// satellite verifies that it signed the order limit
if err := signing.VerifyOrderLimitSignature(ctx, endpoint.satelliteSignee, orderLimit); err != nil {
log.Debug("invalid settlement: unable to verify order limit")
mon.Event("order_not_valid_satellite_signature")
return false
}
// satellite verifies that the order signature matches pub key in order limit
if err := signing.VerifyUplinkOrderSignature(ctx, orderLimit.UplinkPublicKey, order); err != nil {
log.Debug("invalid settlement: unable to verify order")
mon.Event("order_not_valid_uplink_signature")
return false
}
if orderLimit.SerialNumber != order.SerialNumber {
log.Debug("invalid settlement: invalid serial number")
mon.Event("order_not_valid_serialnum_mismatch")
return false
}
// verify the 1 hr windows match
if window != date.TruncateToHourInNano(orderLimit.OrderCreation) {
log.Debug("invalid settlement: window mismatch")
mon.Event("order_not_valid_window_mismatch")
return false
}
if orderLimit.Limit < order.Amount {
log.Debug("invalid settlement: amounts mismatch")
mon.Event("order_not_valid_amounts_mismatch")
return false
}
return true
}