2019-09-25 18:12:44 +01:00
// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package satellitedb
import (
"bytes"
"context"
2019-10-11 22:18:05 +01:00
"database/sql"
2020-07-14 14:04:38 +01:00
"errors"
2019-09-25 18:12:44 +01:00
"sort"
"time"
2019-10-11 22:18:05 +01:00
"github.com/zeebo/errs"
2019-09-25 18:12:44 +01:00
2019-12-27 11:48:47 +00:00
"storj.io/common/storj"
2021-06-25 08:23:33 +01:00
"storj.io/common/uuid"
2021-02-10 18:09:49 +00:00
"storj.io/private/dbutil"
2021-04-23 10:52:40 +01:00
"storj.io/private/dbutil/pgutil"
"storj.io/private/tagsql"
2019-09-25 18:12:44 +01:00
"storj.io/storj/satellite/gracefulexit"
2021-04-21 13:42:57 +01:00
"storj.io/storj/satellite/metabase"
2020-01-15 02:29:51 +00:00
"storj.io/storj/satellite/satellitedb/dbx"
2019-09-25 18:12:44 +01:00
)
type gracefulexitDB struct {
2019-12-14 02:29:54 +00:00
db * satelliteDB
2019-09-25 18:12:44 +01:00
}
2021-02-17 01:25:25 +00:00
const (
deleteExitProgressBatchSize = 1000
)
2019-09-25 18:12:44 +01:00
// IncrementProgress increments transfer stats for a node.
func ( db * gracefulexitDB ) IncrementProgress ( ctx context . Context , nodeID storj . NodeID , bytes int64 , successfulTransfers int64 , failedTransfers int64 ) ( err error ) {
defer mon . Task ( ) ( & ctx ) ( & err )
statement := db . db . Rebind (
2021-10-20 12:14:12 +01:00
` INSERT INTO graceful_exit_progress ( node_id , bytes_transferred , pieces_transferred , pieces_failed , updated_at ) VALUES ( ? , ? , ? , ? , ? )
2019-09-25 18:12:44 +01:00
ON CONFLICT ( node_id )
2021-01-15 15:34:41 +00:00
DO UPDATE SET bytes_transferred = graceful_exit_progress . bytes_transferred + excluded . bytes_transferred ,
2019-09-25 18:12:44 +01:00
pieces_transferred = graceful_exit_progress . pieces_transferred + excluded . pieces_transferred ,
pieces_failed = graceful_exit_progress . pieces_failed + excluded . pieces_failed ,
updated_at = excluded . updated_at ; ` ,
)
now := time . Now ( ) . UTC ( )
_ , err = db . db . ExecContext ( ctx , statement , nodeID , bytes , successfulTransfers , failedTransfers , now )
if err != nil {
return Error . Wrap ( err )
}
return nil
}
// GetProgress gets a graceful exit progress entry.
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 ( ) ) )
2020-07-14 14:04:38 +01:00
if errors . Is ( err , sql . ErrNoRows ) {
2019-10-23 02:06:01 +01:00
return nil , gracefulexit . ErrNodeNotFound . Wrap ( err )
} else if err != nil {
2019-09-25 18:12:44 +01:00
return nil , Error . Wrap ( err )
}
nID , err := storj . NodeIDFromBytes ( dbxProgress . NodeId )
if err != nil {
return nil , Error . Wrap ( err )
}
progress := & gracefulexit . Progress {
2021-09-05 22:29:22 +01:00
NodeID : nID ,
BytesTransferred : dbxProgress . BytesTransferred ,
PiecesTransferred : dbxProgress . PiecesTransferred ,
PiecesFailed : dbxProgress . PiecesFailed ,
UpdatedAt : dbxProgress . UpdatedAt ,
2019-09-25 18:12:44 +01:00
}
return progress , Error . Wrap ( err )
}
2021-02-10 18:09:49 +00:00
// Enqueue batch inserts graceful exit transfer queue entries if it does not exist.
2021-09-05 22:29:22 +01:00
func ( db * gracefulexitDB ) Enqueue ( ctx context . Context , items [ ] gracefulexit . TransferQueueItem , batchSize int ) ( err error ) {
2019-09-25 18:12:44 +01:00
defer mon . Task ( ) ( & ctx ) ( & err )
2021-09-05 22:29:22 +01:00
sort . Slice ( items , func ( i , k int ) bool {
compare := bytes . Compare ( items [ i ] . NodeID . Bytes ( ) , items [ k ] . NodeID . Bytes ( ) )
if compare == 0 {
compare = bytes . Compare ( items [ i ] . StreamID [ : ] , items [ k ] . StreamID [ : ] )
2021-06-25 08:23:33 +01:00
if compare == 0 {
2021-09-05 22:29:22 +01:00
return items [ i ] . Position . Encode ( ) < items [ k ] . Position . Encode ( )
2021-06-25 08:23:33 +01:00
}
return compare < 0
2021-09-05 22:29:22 +01:00
}
return compare < 0
} )
2019-10-18 22:27:57 +01:00
2021-02-10 18:09:49 +00:00
for i := 0 ; i < len ( items ) ; i += batchSize {
lowerBound := i
upperBound := lowerBound + batchSize
if upperBound > len ( items ) {
upperBound = len ( items )
}
var nodeIDs [ ] storj . NodeID
2021-06-25 08:23:33 +01:00
var streamIds [ ] [ ] byte
var positions [ ] int64
2021-02-10 18:09:49 +00:00
var pieceNums [ ] int32
var rootPieceIDs [ ] [ ] byte
var durabilities [ ] float64
2021-06-25 08:23:33 +01:00
2021-02-10 18:09:49 +00:00
for _ , item := range items [ lowerBound : upperBound ] {
2021-06-25 08:23:33 +01:00
item := item
2021-02-10 18:09:49 +00:00
nodeIDs = append ( nodeIDs , item . NodeID )
2021-09-05 22:29:22 +01:00
streamIds = append ( streamIds , item . StreamID [ : ] )
positions = append ( positions , int64 ( item . Position . Encode ( ) ) )
2021-02-10 18:09:49 +00:00
pieceNums = append ( pieceNums , item . PieceNum )
rootPieceIDs = append ( rootPieceIDs , item . RootPieceID . Bytes ( ) )
durabilities = append ( durabilities , item . DurabilityRatio )
}
2019-09-25 18:12:44 +01:00
2021-02-10 18:09:49 +00:00
_ , err = db . db . ExecContext ( ctx , db . db . Rebind ( `
2021-06-25 08:23:33 +01:00
INSERT INTO graceful_exit_segment_transfer_queue (
node_id , stream_id , position , piece_num ,
root_piece_id , durability_ratio , queued_at
) SELECT
unnest ( $ 1 : : bytea [ ] ) , unnest ( $ 2 : : bytea [ ] ) , unnest ( $ 3 : : int8 [ ] ) ,
unnest ( $ 4 : : int4 [ ] ) , unnest ( $ 5 : : bytea [ ] ) , unnest ( $ 6 : : float8 [ ] ) ,
$ 7
ON CONFLICT DO NOTHING ; ` ) , pgutil . NodeIDArray ( nodeIDs ) , pgutil . ByteaArray ( streamIds ) , pgutil . Int8Array ( positions ) ,
pgutil . Int4Array ( pieceNums ) , pgutil . ByteaArray ( rootPieceIDs ) , pgutil . Float8Array ( durabilities ) ,
time . Now ( ) . UTC ( ) )
2019-09-25 18:12:44 +01:00
2021-02-10 18:09:49 +00:00
if err != nil {
return Error . Wrap ( err )
}
2021-06-25 08:23:33 +01:00
}
2021-02-10 18:09:49 +00:00
return nil
2019-09-25 18:12:44 +01:00
}
// UpdateTransferQueueItem creates a graceful exit transfer queue entry.
2021-09-05 22:29:22 +01:00
func ( db * gracefulexitDB ) UpdateTransferQueueItem ( ctx context . Context , item gracefulexit . TransferQueueItem ) ( err error ) {
2019-09-25 18:12:44 +01:00
defer mon . Task ( ) ( & ctx ) ( & err )
2021-06-25 08:23:33 +01:00
update := dbx . GracefulExitSegmentTransfer_Update_Fields {
DurabilityRatio : dbx . GracefulExitSegmentTransfer_DurabilityRatio ( item . DurabilityRatio ) ,
LastFailedCode : dbx . GracefulExitSegmentTransfer_LastFailedCode_Raw ( item . LastFailedCode ) ,
FailedCount : dbx . GracefulExitSegmentTransfer_FailedCount_Raw ( item . FailedCount ) ,
2019-09-25 18:12:44 +01:00
}
2019-10-11 22:18:05 +01:00
if item . RequestedAt != nil {
2021-06-25 08:23:33 +01:00
update . RequestedAt = dbx . GracefulExitSegmentTransfer_RequestedAt_Raw ( item . RequestedAt )
2019-09-25 18:12:44 +01:00
}
2019-10-11 22:18:05 +01:00
if item . LastFailedAt != nil {
2021-06-25 08:23:33 +01:00
update . LastFailedAt = dbx . GracefulExitSegmentTransfer_LastFailedAt_Raw ( item . LastFailedAt )
2019-09-25 18:12:44 +01:00
}
2019-10-11 22:18:05 +01:00
if item . FinishedAt != nil {
2021-06-25 08:23:33 +01:00
update . FinishedAt = dbx . GracefulExitSegmentTransfer_FinishedAt_Raw ( item . FinishedAt )
2019-09-25 18:12:44 +01:00
}
2021-06-25 08:23:33 +01:00
return db . db . UpdateNoReturn_GracefulExitSegmentTransfer_By_NodeId_And_StreamId_And_Position_And_PieceNum ( ctx ,
dbx . GracefulExitSegmentTransfer_NodeId ( item . NodeID . Bytes ( ) ) ,
dbx . GracefulExitSegmentTransfer_StreamId ( item . StreamID [ : ] ) ,
dbx . GracefulExitSegmentTransfer_Position ( item . Position . Encode ( ) ) ,
dbx . GracefulExitSegmentTransfer_PieceNum ( int ( item . PieceNum ) ) ,
2019-09-25 18:12:44 +01:00
update ,
)
}
// DeleteTransferQueueItem deletes a graceful exit transfer queue entry.
2021-09-05 22:29:22 +01:00
func ( db * gracefulexitDB ) DeleteTransferQueueItem ( ctx context . Context , nodeID storj . NodeID , streamID uuid . UUID , position metabase . SegmentPosition , pieceNum int32 ) ( err error ) {
2019-09-25 18:12:44 +01:00
defer mon . Task ( ) ( & ctx ) ( & err )
2021-06-25 08:23:33 +01:00
_ , err = db . db . Delete_GracefulExitSegmentTransfer_By_NodeId_And_StreamId_And_Position_And_PieceNum ( ctx ,
dbx . GracefulExitSegmentTransfer_NodeId ( nodeID . Bytes ( ) ) ,
dbx . GracefulExitSegmentTransfer_StreamId ( streamID [ : ] ) ,
dbx . GracefulExitSegmentTransfer_Position ( position . Encode ( ) ) , dbx . GracefulExitSegmentTransfer_PieceNum ( int ( pieceNum ) ) )
2019-09-25 18:12:44 +01:00
return Error . Wrap ( err )
}
// DeleteTransferQueueItem deletes a graceful exit transfer queue entries by nodeID.
2021-09-05 22:29:22 +01:00
func ( db * gracefulexitDB ) DeleteTransferQueueItems ( ctx context . Context , nodeID storj . NodeID ) ( err error ) {
2019-09-25 18:12:44 +01:00
defer mon . Task ( ) ( & ctx ) ( & err )
2021-09-05 22:29:22 +01:00
2021-06-25 08:23:33 +01:00
_ , err = db . db . Delete_GracefulExitSegmentTransfer_By_NodeId ( ctx , dbx . GracefulExitSegmentTransfer_NodeId ( nodeID . Bytes ( ) ) )
2019-09-25 18:12:44 +01:00
return Error . Wrap ( err )
2021-06-25 08:23:33 +01:00
2019-09-25 18:12:44 +01:00
}
2021-01-15 15:34:41 +00:00
// DeleteFinishedTransferQueueItem deletes finished graceful exit transfer queue entries by nodeID.
2021-09-05 22:29:22 +01:00
func ( db * gracefulexitDB ) DeleteFinishedTransferQueueItems ( ctx context . Context , nodeID storj . NodeID ) ( err error ) {
2019-09-25 18:12:44 +01:00
defer mon . Task ( ) ( & ctx ) ( & err )
2021-06-25 08:23:33 +01:00
_ , err = db . db . Delete_GracefulExitSegmentTransfer_By_NodeId_And_FinishedAt_IsNot_Null ( ctx , dbx . GracefulExitSegmentTransfer_NodeId ( nodeID . Bytes ( ) ) )
2019-09-25 18:12:44 +01:00
return Error . Wrap ( err )
}
2021-01-14 15:57:04 +00:00
// DeleteAllFinishedTransferQueueItems deletes all graceful exit transfer
// queue items whose nodes have finished the exit before the indicated time
// returning the total number of deleted items.
func ( db * gracefulexitDB ) DeleteAllFinishedTransferQueueItems (
2021-05-11 09:49:26 +01:00
ctx context . Context , before time . Time , asOfSystemInterval time . Duration , batchSize int ) ( _ int64 , err error ) {
2021-01-14 15:57:04 +00:00
defer mon . Task ( ) ( & ctx ) ( & err )
2021-05-11 09:49:26 +01:00
switch db . db . impl {
2021-02-10 18:09:49 +00:00
case dbutil . Postgres :
statement := `
2021-06-25 08:23:33 +01:00
DELETE FROM graceful_exit_segment_transfer_queue
2021-02-10 18:09:49 +00:00
WHERE node_id IN (
2021-06-25 08:23:33 +01:00
SELECT node_id FROM graceful_exit_segment_transfer_queue INNER JOIN nodes
ON graceful_exit_segment_transfer_queue . node_id = nodes . id
WHERE nodes . exit_finished_at IS NOT NULL
AND nodes . exit_finished_at < $ 1
2021-02-10 18:09:49 +00:00
) `
res , err := db . db . ExecContext ( ctx , statement , before )
if err != nil {
return 0 , Error . Wrap ( err )
}
2021-01-14 15:57:04 +00:00
2021-02-10 18:09:49 +00:00
count , err := res . RowsAffected ( )
if err != nil {
return 0 , Error . Wrap ( err )
}
2021-01-14 15:57:04 +00:00
2021-09-05 22:29:22 +01:00
return count , nil
2021-02-10 18:09:49 +00:00
case dbutil . Cockroach :
nodesQuery := `
SELECT id
2021-05-11 09:49:26 +01:00
FROM nodes
` + db.db.impl.AsOfSystemInterval(asOfSystemInterval) + `
2021-02-10 18:09:49 +00:00
WHERE exit_finished_at IS NOT NULL
AND exit_finished_at < $ 1
LIMIT $ 2 OFFSET $ 3
`
deleteStmt := `
2021-06-25 08:23:33 +01:00
DELETE FROM graceful_exit_segment_transfer_queue
2021-02-10 18:09:49 +00:00
WHERE node_id = $ 1
LIMIT $ 2
`
2021-09-05 22:29:22 +01:00
2021-02-10 18:09:49 +00:00
var (
deleteCount int64
offset int
)
for {
var nodeIDs storj . NodeIDList
deleteItems := func ( ) ( int64 , error ) {
// Select exited nodes
rows , err := db . db . QueryContext ( ctx , nodesQuery , before , batchSize , offset )
if err != nil {
return deleteCount , Error . Wrap ( err )
}
defer func ( ) { err = errs . Combine ( err , rows . Close ( ) ) } ( )
count := 0
for rows . Next ( ) {
var id storj . NodeID
if err = rows . Scan ( & id ) ; err != nil {
return deleteCount , Error . Wrap ( err )
}
nodeIDs = append ( nodeIDs , id )
count ++
}
if count == batchSize {
offset += count
} else {
offset = - 1 // indicates that there aren't more nodes to query
}
for _ , id := range nodeIDs {
for {
res , err := db . db . ExecContext ( ctx , deleteStmt , id . Bytes ( ) , batchSize )
if err != nil {
return deleteCount , Error . Wrap ( err )
}
count , err := res . RowsAffected ( )
if err != nil {
return deleteCount , Error . Wrap ( err )
}
deleteCount += count
if count < int64 ( batchSize ) {
break
}
}
}
return deleteCount , nil
}
deleteCount , err = deleteItems ( )
if err != nil {
return deleteCount , err
}
// when offset is negative means that we have get already all the nodes
// which have exited
if offset < 0 {
break
}
}
return deleteCount , nil
2021-01-14 15:57:04 +00:00
}
2021-05-11 09:49:26 +01:00
return 0 , Error . New ( "unsupported implementation: %s" , db . db . impl )
2021-01-14 15:57:04 +00:00
}
2021-02-17 01:25:25 +00:00
// DeleteFinishedExitProgress deletes exit progress entries for nodes that
// finished exiting before the indicated time, returns number of deleted entries.
func ( db * gracefulexitDB ) DeleteFinishedExitProgress (
2021-05-11 09:49:26 +01:00
ctx context . Context , before time . Time , asOfSystemInterval time . Duration ) ( _ int64 , err error ) {
2021-02-17 01:25:25 +00:00
defer mon . Task ( ) ( & ctx ) ( & err )
2021-05-11 09:49:26 +01:00
finishedNodes , err := db . GetFinishedExitNodes ( ctx , before , asOfSystemInterval )
2021-02-17 01:25:25 +00:00
if err != nil {
return 0 , err
}
return db . DeleteBatchExitProgress ( ctx , finishedNodes )
}
// GetFinishedExitNodes gets nodes that are marked having finished graceful exit before a given time.
2021-05-11 09:49:26 +01:00
func ( db * gracefulexitDB ) GetFinishedExitNodes ( ctx context . Context , before time . Time , asOfSystemInterval time . Duration ) ( finishedNodes [ ] storj . NodeID , err error ) {
2021-02-17 01:25:25 +00:00
defer mon . Task ( ) ( & ctx ) ( & err )
2021-05-11 09:49:26 +01:00
stmt := `
SELECT id
FROM nodes
` + db.db.impl.AsOfSystemInterval(asOfSystemInterval) + `
2021-02-10 18:09:49 +00:00
WHERE exit_finished_at IS NOT NULL
AND exit_finished_at < ?
`
rows , err := db . db . Query ( ctx , db . db . Rebind ( stmt ) , before . UTC ( ) )
2021-02-17 01:25:25 +00:00
if err != nil {
return nil , Error . Wrap ( err )
}
defer func ( ) {
err = Error . Wrap ( errs . Combine ( err , rows . Close ( ) ) )
} ( )
for rows . Next ( ) {
var id storj . NodeID
err = rows . Scan ( & id )
if err != nil {
return nil , Error . Wrap ( err )
}
finishedNodes = append ( finishedNodes , id )
}
return finishedNodes , Error . Wrap ( rows . Err ( ) )
}
// DeleteBatchExitProgress batch deletes from exit progress. This is separate from
// getting the node IDs because the combined query is slow in CRDB. It's safe to do
// separately because if nodes are deleted between the get and delete, it doesn't
// affect correctness.
func ( db * gracefulexitDB ) DeleteBatchExitProgress ( ctx context . Context , nodeIDs [ ] storj . NodeID ) ( deleted int64 , err error ) {
defer mon . Task ( ) ( & ctx ) ( & err )
2022-02-22 14:41:20 +00:00
stmt := ` DELETE FROM graceful_exit_progress
2021-02-17 01:25:25 +00:00
WHERE node_id = ANY ( $ 1 ) `
for len ( nodeIDs ) > 0 {
numToSubmit := len ( nodeIDs )
if numToSubmit > deleteExitProgressBatchSize {
numToSubmit = deleteExitProgressBatchSize
}
nodesToSubmit := nodeIDs [ : numToSubmit ]
res , err := db . db . ExecContext ( ctx , stmt , pgutil . NodeIDArray ( nodesToSubmit ) )
if err != nil {
return deleted , Error . Wrap ( err )
}
count , err := res . RowsAffected ( )
if err != nil {
return deleted , Error . Wrap ( err )
}
deleted += count
nodeIDs = nodeIDs [ numToSubmit : ]
}
return deleted , Error . Wrap ( err )
}
2019-09-25 18:12:44 +01:00
// GetTransferQueueItem gets a graceful exit transfer queue entry.
2021-09-05 22:29:22 +01:00
func ( db * gracefulexitDB ) GetTransferQueueItem ( ctx context . Context , nodeID storj . NodeID , streamID uuid . UUID , position metabase . SegmentPosition , pieceNum int32 ) ( _ * gracefulexit . TransferQueueItem , err error ) {
2019-09-25 18:12:44 +01:00
defer mon . Task ( ) ( & ctx ) ( & err )
2021-06-25 08:23:33 +01:00
dbxTransferQueue , err := db . db . Get_GracefulExitSegmentTransfer_By_NodeId_And_StreamId_And_Position_And_PieceNum ( ctx ,
dbx . GracefulExitSegmentTransfer_NodeId ( nodeID . Bytes ( ) ) ,
dbx . GracefulExitSegmentTransfer_StreamId ( streamID [ : ] ) ,
dbx . GracefulExitSegmentTransfer_Position ( position . Encode ( ) ) ,
dbx . GracefulExitSegmentTransfer_PieceNum ( int ( pieceNum ) ) )
2019-09-25 18:12:44 +01:00
if err != nil {
return nil , Error . Wrap ( err )
}
2021-06-25 08:23:33 +01:00
transferQueueItem , err := dbxSegmentTransferToTransferQueueItem ( dbxTransferQueue )
2019-09-25 18:12:44 +01:00
if err != nil {
return nil , Error . Wrap ( err )
}
return transferQueueItem , Error . Wrap ( err )
2021-06-25 08:23:33 +01:00
2019-09-25 18:12:44 +01:00
}
2019-10-11 22:18:05 +01:00
// GetIncomplete gets incomplete graceful exit transfer queue entries ordered by durability ratio and queued date ascending.
2021-09-05 22:29:22 +01:00
func ( db * gracefulexitDB ) GetIncomplete ( ctx context . Context , nodeID storj . NodeID , limit int , offset int64 ) ( _ [ ] * gracefulexit . TransferQueueItem , err error ) {
2019-09-25 18:12:44 +01:00
defer mon . Task ( ) ( & ctx ) ( & err )
2021-06-25 08:23:33 +01:00
2021-09-05 22:29:22 +01:00
sql := `
2021-06-25 08:23:33 +01:00
SELECT
node_id , stream_id , position ,
piece_num , root_piece_id , durability_ratio ,
queued_at , requested_at , last_failed_at ,
last_failed_code , failed_count , finished_at ,
order_limit_send_count
FROM graceful_exit_segment_transfer_queue
WHERE node_id = ?
AND finished_at is NULL
ORDER BY durability_ratio asc , queued_at asc LIMIT ? OFFSET ? `
2020-01-17 20:07:00 +00:00
rows , err := db . db . Query ( ctx , db . db . Rebind ( sql ) , nodeID . Bytes ( ) , limit , offset )
2019-09-25 18:12:44 +01:00
if err != nil {
return nil , Error . Wrap ( err )
}
2020-01-16 14:27:24 +00:00
defer func ( ) { err = errs . Combine ( err , rows . Close ( ) ) } ( )
2019-10-11 22:18:05 +01:00
2021-09-05 22:29:22 +01:00
transferQueueItemRows , err := scanRows ( rows )
2019-10-11 22:18:05 +01:00
if err != nil {
return nil , Error . Wrap ( err )
}
return transferQueueItemRows , nil
}
// GetIncompleteNotFailed gets incomplete graceful exit transfer queue entries that haven't failed, ordered by durability ratio and queued date ascending.
2021-09-05 22:29:22 +01:00
func ( db * gracefulexitDB ) GetIncompleteNotFailed ( ctx context . Context , nodeID storj . NodeID , limit int , offset int64 ) ( _ [ ] * gracefulexit . TransferQueueItem , err error ) {
2019-10-11 22:18:05 +01:00
defer mon . Task ( ) ( & ctx ) ( & err )
2021-06-25 08:23:33 +01:00
2021-09-05 22:29:22 +01:00
sql := `
2021-06-25 08:23:33 +01:00
SELECT
node_id , stream_id , position ,
piece_num , root_piece_id , durability_ratio ,
queued_at , requested_at , last_failed_at ,
last_failed_code , failed_count , finished_at ,
order_limit_send_count
FROM graceful_exit_segment_transfer_queue
WHERE node_id = ?
AND finished_at is NULL
AND last_failed_at is NULL
ORDER BY durability_ratio asc , queued_at asc LIMIT ? OFFSET ? `
2020-01-17 20:07:00 +00:00
rows , err := db . db . Query ( ctx , db . db . Rebind ( sql ) , nodeID . Bytes ( ) , limit , offset )
2019-10-11 22:18:05 +01:00
if err != nil {
return nil , Error . Wrap ( err )
}
2020-01-16 14:27:24 +00:00
defer func ( ) { err = errs . Combine ( err , rows . Close ( ) ) } ( )
2019-10-11 22:18:05 +01:00
2021-09-05 22:29:22 +01:00
transferQueueItemRows , err := scanRows ( rows )
2019-10-11 22:18:05 +01:00
if err != nil {
return nil , Error . Wrap ( err )
}
return transferQueueItemRows , nil
}
// GetIncompleteNotFailed gets incomplete graceful exit transfer queue entries that have failed <= maxFailures times, ordered by durability ratio and queued date ascending.
2021-09-05 22:29:22 +01:00
func ( db * gracefulexitDB ) GetIncompleteFailed ( ctx context . Context , nodeID storj . NodeID , maxFailures int , limit int , offset int64 ) ( _ [ ] * gracefulexit . TransferQueueItem , err error ) {
2019-10-11 22:18:05 +01:00
defer mon . Task ( ) ( & ctx ) ( & err )
2021-06-25 08:23:33 +01:00
2021-09-05 22:29:22 +01:00
sql := `
2021-06-25 08:23:33 +01:00
SELECT
node_id , stream_id , position ,
piece_num , root_piece_id , durability_ratio ,
queued_at , requested_at , last_failed_at ,
last_failed_code , failed_count , finished_at ,
order_limit_send_count
FROM graceful_exit_segment_transfer_queue
WHERE node_id = ?
2022-02-22 14:41:20 +00:00
AND finished_at IS NULL
AND last_failed_at IS NOT NULL
2021-06-25 08:23:33 +01:00
AND failed_count < ?
2019-10-11 22:18:05 +01:00
ORDER BY durability_ratio asc , queued_at asc LIMIT ? OFFSET ? `
2020-01-17 20:07:00 +00:00
rows , err := db . db . Query ( ctx , db . db . Rebind ( sql ) , nodeID . Bytes ( ) , maxFailures , limit , offset )
2019-10-11 22:18:05 +01:00
if err != nil {
return nil , Error . Wrap ( err )
}
2020-01-16 14:27:24 +00:00
defer func ( ) { err = errs . Combine ( err , rows . Close ( ) ) } ( )
2019-10-11 22:18:05 +01:00
2021-09-05 22:29:22 +01:00
transferQueueItemRows , err := scanRows ( rows )
2019-10-11 22:18:05 +01:00
if err != nil {
return nil , Error . Wrap ( err )
}
return transferQueueItemRows , nil
}
2019-11-13 14:54:50 +00:00
// IncrementOrderLimitSendCount increments the number of times a node has been sent an order limit for transferring.
2021-09-05 22:29:22 +01:00
func ( db * gracefulexitDB ) IncrementOrderLimitSendCount ( ctx context . Context , nodeID storj . NodeID , streamID uuid . UUID , position metabase . SegmentPosition , pieceNum int32 ) ( err error ) {
2019-11-13 14:54:50 +00:00
defer mon . Task ( ) ( & ctx ) ( & err )
2021-09-05 22:29:22 +01:00
sql := ` UPDATE graceful_exit_segment_transfer_queue SET order_limit_send_count = graceful_exit_segment_transfer_queue . order_limit_send_count + 1
2021-06-25 08:23:33 +01:00
WHERE node_id = ?
AND stream_id = ?
AND position = ?
AND piece_num = ? `
2021-09-05 22:29:22 +01:00
_ , err = db . db . ExecContext ( ctx , db . db . Rebind ( sql ) , nodeID , streamID , position . Encode ( ) , pieceNum )
2021-06-25 08:23:33 +01:00
2020-01-19 14:59:49 +00:00
return Error . Wrap ( err )
2019-11-13 14:54:50 +00:00
}
2021-01-14 15:57:04 +00:00
// CountFinishedTransferQueueItemsByNode return a map of the nodes which has
// finished the exit before the indicated time but there are at least one item
// left in the transfer queue.
2021-05-11 09:49:26 +01:00
func ( db * gracefulexitDB ) CountFinishedTransferQueueItemsByNode ( ctx context . Context , before time . Time , asOfSystemInterval time . Duration ) ( _ map [ storj . NodeID ] int64 , err error ) {
2021-01-14 15:57:04 +00:00
defer mon . Task ( ) ( & ctx ) ( & err )
2021-02-10 18:09:49 +00:00
query := ` SELECT n . id , count ( getq . node_id )
2021-06-25 08:23:33 +01:00
FROM nodes as n INNER JOIN graceful_exit_segment_transfer_queue as getq
2021-05-11 09:49:26 +01:00
ON n . id = getq . node_id
` + db.db.impl.AsOfSystemInterval(asOfSystemInterval) + `
2021-01-14 15:57:04 +00:00
WHERE n . exit_finished_at IS NOT NULL
AND n . exit_finished_at < ?
2021-02-10 18:09:49 +00:00
GROUP BY n . id `
statement := db . db . Rebind ( query )
2021-01-14 15:57:04 +00:00
rows , err := db . db . QueryContext ( ctx , statement , before )
if err != nil {
return nil , Error . Wrap ( err )
}
defer func ( ) { err = errs . Combine ( err , Error . Wrap ( rows . Close ( ) ) ) } ( )
nodesItemsCount := make ( map [ storj . NodeID ] int64 )
for rows . Next ( ) {
var (
nodeID storj . NodeID
n int64
)
err := rows . Scan ( & nodeID , & n )
if err != nil {
return nil , Error . Wrap ( err )
}
nodesItemsCount [ nodeID ] = n
}
2021-09-05 22:29:22 +01:00
return nodesItemsCount , Error . Wrap ( rows . Err ( ) )
2021-01-14 15:57:04 +00:00
}
2021-09-05 22:29:22 +01:00
func scanRows ( rows tagsql . Rows ) ( transferQueueItemRows [ ] * gracefulexit . TransferQueueItem , err error ) {
2019-10-11 22:18:05 +01:00
for rows . Next ( ) {
transferQueueItem := & gracefulexit . TransferQueueItem { }
2019-11-07 16:13:05 +00:00
var pieceIDBytes [ ] byte
2021-09-05 22:29:22 +01:00
err = rows . Scan ( & transferQueueItem . NodeID , & transferQueueItem . StreamID , & transferQueueItem . Position , & transferQueueItem . PieceNum , & pieceIDBytes ,
& transferQueueItem . DurabilityRatio , & transferQueueItem . QueuedAt , & transferQueueItem . RequestedAt , & transferQueueItem . LastFailedAt ,
& transferQueueItem . LastFailedCode , & transferQueueItem . FailedCount , & transferQueueItem . FinishedAt , & transferQueueItem . OrderLimitSendCount )
2021-06-25 08:23:33 +01:00
2019-09-25 18:12:44 +01:00
if err != nil {
return nil , Error . Wrap ( err )
}
2019-11-07 16:13:05 +00:00
if pieceIDBytes != nil {
transferQueueItem . RootPieceID , err = storj . PieceIDFromBytes ( pieceIDBytes )
if err != nil {
return nil , Error . Wrap ( err )
}
}
2019-09-25 18:12:44 +01:00
2019-10-11 22:18:05 +01:00
transferQueueItemRows = append ( transferQueueItemRows , transferQueueItem )
}
2020-01-19 14:59:49 +00:00
return transferQueueItemRows , Error . Wrap ( rows . Err ( ) )
2019-09-25 18:12:44 +01:00
}
2021-06-25 08:23:33 +01:00
func dbxSegmentTransferToTransferQueueItem ( dbxSegmentTransfer * dbx . GracefulExitSegmentTransfer ) ( item * gracefulexit . TransferQueueItem , err error ) {
nID , err := storj . NodeIDFromBytes ( dbxSegmentTransfer . NodeId )
if err != nil {
return nil , Error . Wrap ( err )
}
streamID , err := uuid . FromBytes ( dbxSegmentTransfer . StreamId )
if err != nil {
return nil , Error . Wrap ( err )
}
position := metabase . SegmentPositionFromEncoded ( dbxSegmentTransfer . Position )
item = & gracefulexit . TransferQueueItem {
NodeID : nID ,
StreamID : streamID ,
Position : position ,
PieceNum : int32 ( dbxSegmentTransfer . PieceNum ) ,
DurabilityRatio : dbxSegmentTransfer . DurabilityRatio ,
QueuedAt : dbxSegmentTransfer . QueuedAt ,
OrderLimitSendCount : dbxSegmentTransfer . OrderLimitSendCount ,
}
if dbxSegmentTransfer . RootPieceId != nil {
item . RootPieceID , err = storj . PieceIDFromBytes ( dbxSegmentTransfer . RootPieceId )
if err != nil {
return nil , err
}
}
if dbxSegmentTransfer . LastFailedCode != nil {
item . LastFailedCode = dbxSegmentTransfer . LastFailedCode
}
if dbxSegmentTransfer . FailedCount != nil {
item . FailedCount = dbxSegmentTransfer . FailedCount
}
if dbxSegmentTransfer . RequestedAt != nil && ! dbxSegmentTransfer . RequestedAt . IsZero ( ) {
item . RequestedAt = dbxSegmentTransfer . RequestedAt
}
if dbxSegmentTransfer . LastFailedAt != nil && ! dbxSegmentTransfer . LastFailedAt . IsZero ( ) {
item . LastFailedAt = dbxSegmentTransfer . LastFailedAt
}
if dbxSegmentTransfer . FinishedAt != nil && ! dbxSegmentTransfer . FinishedAt . IsZero ( ) {
item . FinishedAt = dbxSegmentTransfer . FinishedAt
}
return item , nil
}