420 lines
11 KiB
Go
420 lines
11 KiB
Go
// Copyright (C) 2021 Storj Labs, Inc.
|
|
// See LICENSE for copying information.
|
|
|
|
package metabase
|
|
|
|
import (
|
|
"context"
|
|
"math"
|
|
"time"
|
|
|
|
"github.com/zeebo/errs"
|
|
|
|
"storj.io/common/storj"
|
|
"storj.io/common/uuid"
|
|
"storj.io/private/tagsql"
|
|
)
|
|
|
|
const loopIteratorBatchSizeLimit = intLimitRange(5000)
|
|
|
|
// IterateLoopObjects contains arguments necessary for listing objects in metabase.
|
|
type IterateLoopObjects struct {
|
|
BatchSize int
|
|
|
|
AsOfSystemTime time.Time
|
|
AsOfSystemInterval time.Duration
|
|
}
|
|
|
|
// Verify verifies get object request fields.
|
|
func (opts *IterateLoopObjects) Verify() error {
|
|
if opts.BatchSize < 0 {
|
|
return ErrInvalidRequest.New("BatchSize is negative")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// LoopObjectsIterator iterates over a sequence of LoopObjectEntry items.
|
|
type LoopObjectsIterator interface {
|
|
Next(ctx context.Context, item *LoopObjectEntry) bool
|
|
}
|
|
|
|
// LoopObjectEntry contains information about object needed by metainfo loop.
|
|
type LoopObjectEntry struct {
|
|
ObjectStream // metrics, repair, tally
|
|
Status ObjectStatus // verify
|
|
CreatedAt time.Time // temp used by metabase-createdat-migration
|
|
ExpiresAt *time.Time // tally
|
|
SegmentCount int32 // metrics
|
|
TotalEncryptedSize int64 // tally
|
|
EncryptedMetadataSize int // tally
|
|
}
|
|
|
|
// Expired checks if object is expired relative to now.
|
|
func (o LoopObjectEntry) Expired(now time.Time) bool {
|
|
return o.ExpiresAt != nil && o.ExpiresAt.Before(now)
|
|
}
|
|
|
|
// IterateLoopObjects iterates through all objects in metabase.
|
|
func (db *DB) IterateLoopObjects(ctx context.Context, opts IterateLoopObjects, fn func(context.Context, LoopObjectsIterator) error) (err error) {
|
|
defer mon.Task()(&ctx)(&err)
|
|
|
|
if err := opts.Verify(); err != nil {
|
|
return err
|
|
}
|
|
|
|
it := &loopIterator{
|
|
db: db,
|
|
|
|
batchSize: opts.BatchSize,
|
|
|
|
curIndex: 0,
|
|
cursor: loopIterateCursor{},
|
|
asOfSystemTime: opts.AsOfSystemTime,
|
|
asOfSystemInterval: opts.AsOfSystemInterval,
|
|
}
|
|
|
|
loopIteratorBatchSizeLimit.Ensure(&it.batchSize)
|
|
|
|
it.curRows, err = it.doNextQuery(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
defer func() {
|
|
if rowsErr := it.curRows.Err(); rowsErr != nil {
|
|
err = errs.Combine(err, rowsErr)
|
|
}
|
|
err = errs.Combine(err, it.failErr, it.curRows.Close())
|
|
}()
|
|
|
|
return fn(ctx, it)
|
|
}
|
|
|
|
// loopIterator enables iteration of all objects in metabase.
|
|
type loopIterator struct {
|
|
db *DB
|
|
|
|
batchSize int
|
|
asOfSystemTime time.Time
|
|
asOfSystemInterval time.Duration
|
|
|
|
curIndex int
|
|
curRows tagsql.Rows
|
|
cursor loopIterateCursor
|
|
|
|
// failErr is set when either scan or next query fails during iteration.
|
|
failErr error
|
|
}
|
|
|
|
type loopIterateCursor struct {
|
|
ProjectID uuid.UUID
|
|
BucketName string
|
|
ObjectKey ObjectKey
|
|
Version Version
|
|
}
|
|
|
|
// Next returns true if there was another item and copy it in item.
|
|
func (it *loopIterator) Next(ctx context.Context, item *LoopObjectEntry) bool {
|
|
next := it.curRows.Next()
|
|
if !next {
|
|
if it.curIndex < it.batchSize {
|
|
return false
|
|
}
|
|
|
|
if it.curRows.Err() != nil {
|
|
return false
|
|
}
|
|
|
|
rows, err := it.doNextQuery(ctx)
|
|
if err != nil {
|
|
it.failErr = errs.Combine(it.failErr, err)
|
|
return false
|
|
}
|
|
|
|
if closeErr := it.curRows.Close(); closeErr != nil {
|
|
it.failErr = errs.Combine(it.failErr, closeErr, rows.Close())
|
|
return false
|
|
}
|
|
|
|
it.curRows = rows
|
|
it.curIndex = 0
|
|
if !it.curRows.Next() {
|
|
return false
|
|
}
|
|
}
|
|
|
|
err := it.scanItem(item)
|
|
if err != nil {
|
|
it.failErr = errs.Combine(it.failErr, err)
|
|
return false
|
|
}
|
|
|
|
it.curIndex++
|
|
it.cursor.ProjectID = item.ProjectID
|
|
it.cursor.BucketName = item.BucketName
|
|
it.cursor.ObjectKey = item.ObjectKey
|
|
it.cursor.Version = item.Version
|
|
|
|
return true
|
|
}
|
|
|
|
func (it *loopIterator) doNextQuery(ctx context.Context) (_ tagsql.Rows, err error) {
|
|
defer mon.Task()(&ctx)(&err)
|
|
|
|
return it.db.db.QueryContext(ctx, `
|
|
SELECT
|
|
project_id, bucket_name,
|
|
object_key, stream_id, version,
|
|
status,
|
|
created_at, expires_at,
|
|
segment_count, total_encrypted_size,
|
|
LENGTH(COALESCE(encrypted_metadata,''))
|
|
FROM objects
|
|
`+it.db.asOfTime(it.asOfSystemTime, it.asOfSystemInterval)+`
|
|
WHERE (project_id, bucket_name, object_key, version) > ($1, $2, $3, $4)
|
|
ORDER BY project_id ASC, bucket_name ASC, object_key ASC, version ASC
|
|
LIMIT $5
|
|
`, it.cursor.ProjectID, []byte(it.cursor.BucketName),
|
|
[]byte(it.cursor.ObjectKey), int(it.cursor.Version),
|
|
it.batchSize,
|
|
)
|
|
}
|
|
|
|
// scanItem scans doNextQuery results into LoopObjectEntry.
|
|
func (it *loopIterator) scanItem(item *LoopObjectEntry) error {
|
|
return it.curRows.Scan(
|
|
&item.ProjectID, &item.BucketName,
|
|
&item.ObjectKey, &item.StreamID, &item.Version,
|
|
&item.Status,
|
|
&item.CreatedAt, &item.ExpiresAt,
|
|
&item.SegmentCount, &item.TotalEncryptedSize,
|
|
&item.EncryptedMetadataSize,
|
|
)
|
|
}
|
|
|
|
// SegmentIterator returns the next segment.
|
|
type SegmentIterator func(ctx context.Context, segment *LoopSegmentEntry) bool
|
|
|
|
// LoopSegmentEntry contains information about segment metadata needed by metainfo loop.
|
|
type LoopSegmentEntry struct {
|
|
StreamID uuid.UUID
|
|
Position SegmentPosition
|
|
CreatedAt time.Time // non-nillable
|
|
ExpiresAt *time.Time
|
|
RepairedAt *time.Time // repair
|
|
RootPieceID storj.PieceID
|
|
EncryptedSize int32 // size of the whole segment (not a piece)
|
|
PlainOffset int64 // verify
|
|
PlainSize int32 // verify
|
|
AliasPieces AliasPieces
|
|
Redundancy storj.RedundancyScheme
|
|
Pieces Pieces
|
|
Placement storj.PlacementConstraint
|
|
}
|
|
|
|
// Inline returns true if segment is inline.
|
|
func (s LoopSegmentEntry) Inline() bool {
|
|
return s.Redundancy.IsZero() && len(s.Pieces) == 0
|
|
}
|
|
|
|
// LoopSegmentsIterator iterates over a sequence of LoopSegmentEntry items.
|
|
type LoopSegmentsIterator interface {
|
|
Next(ctx context.Context, item *LoopSegmentEntry) bool
|
|
}
|
|
|
|
// IterateLoopSegments contains arguments necessary for listing segments in metabase.
|
|
type IterateLoopSegments struct {
|
|
BatchSize int
|
|
AsOfSystemTime time.Time
|
|
AsOfSystemInterval time.Duration
|
|
StartStreamID uuid.UUID
|
|
EndStreamID uuid.UUID
|
|
}
|
|
|
|
// Verify verifies segments request fields.
|
|
func (opts *IterateLoopSegments) Verify() error {
|
|
if opts.BatchSize < 0 {
|
|
return ErrInvalidRequest.New("BatchSize is negative")
|
|
}
|
|
if !opts.EndStreamID.IsZero() {
|
|
if opts.EndStreamID.Less(opts.StartStreamID) {
|
|
return ErrInvalidRequest.New("EndStreamID is smaller than StartStreamID")
|
|
}
|
|
if opts.StartStreamID == opts.EndStreamID {
|
|
return ErrInvalidRequest.New("StartStreamID and EndStreamID must be different")
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// IterateLoopSegments iterates through all segments in metabase.
|
|
func (db *DB) IterateLoopSegments(ctx context.Context, opts IterateLoopSegments, fn func(context.Context, LoopSegmentsIterator) error) (err error) {
|
|
defer mon.Task()(&ctx)(&err)
|
|
|
|
if err := opts.Verify(); err != nil {
|
|
return err
|
|
}
|
|
|
|
loopIteratorBatchSizeLimit.Ensure(&opts.BatchSize)
|
|
|
|
it := &loopSegmentIterator{
|
|
db: db,
|
|
|
|
asOfSystemTime: opts.AsOfSystemTime,
|
|
asOfSystemInterval: opts.AsOfSystemInterval,
|
|
batchSize: opts.BatchSize,
|
|
batchPieces: make([]Pieces, opts.BatchSize),
|
|
|
|
curIndex: 0,
|
|
cursor: loopSegmentIteratorCursor{
|
|
StartStreamID: opts.StartStreamID,
|
|
EndStreamID: opts.EndStreamID,
|
|
},
|
|
}
|
|
|
|
if !opts.StartStreamID.IsZero() {
|
|
// uses MaxInt32 instead of MaxUint32 because position is an int8 in db.
|
|
it.cursor.StartPosition = SegmentPosition{math.MaxInt32, math.MaxInt32}
|
|
}
|
|
if it.cursor.EndStreamID.IsZero() {
|
|
it.cursor.EndStreamID = uuid.Max()
|
|
}
|
|
|
|
it.curRows, err = it.doNextQuery(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
defer func() {
|
|
if rowsErr := it.curRows.Err(); rowsErr != nil {
|
|
err = errs.Combine(err, rowsErr)
|
|
}
|
|
err = errs.Combine(err, it.failErr, it.curRows.Close())
|
|
}()
|
|
|
|
return fn(ctx, it)
|
|
}
|
|
|
|
// loopSegmentIterator enables iteration of all segments in metabase.
|
|
type loopSegmentIterator struct {
|
|
db *DB
|
|
|
|
batchSize int
|
|
// batchPieces are reused between result pages to reduce memory consumption
|
|
batchPieces []Pieces
|
|
|
|
asOfSystemTime time.Time
|
|
asOfSystemInterval time.Duration
|
|
|
|
curIndex int
|
|
curRows tagsql.Rows
|
|
cursor loopSegmentIteratorCursor
|
|
|
|
// failErr is set when either scan or next query fails during iteration.
|
|
failErr error
|
|
}
|
|
|
|
type loopSegmentIteratorCursor struct {
|
|
StartStreamID uuid.UUID
|
|
StartPosition SegmentPosition
|
|
EndStreamID uuid.UUID
|
|
}
|
|
|
|
// Next returns true if there was another item and copy it in item.
|
|
func (it *loopSegmentIterator) Next(ctx context.Context, item *LoopSegmentEntry) bool {
|
|
next := it.curRows.Next()
|
|
if !next {
|
|
if it.curIndex < it.batchSize {
|
|
return false
|
|
}
|
|
|
|
if it.curRows.Err() != nil {
|
|
return false
|
|
}
|
|
|
|
rows, err := it.doNextQuery(ctx)
|
|
if err != nil {
|
|
it.failErr = errs.Combine(it.failErr, err)
|
|
return false
|
|
}
|
|
|
|
if failErr := it.curRows.Close(); failErr != nil {
|
|
it.failErr = errs.Combine(it.failErr, failErr, rows.Close())
|
|
return false
|
|
}
|
|
|
|
it.curRows = rows
|
|
it.curIndex = 0
|
|
if !it.curRows.Next() {
|
|
return false
|
|
}
|
|
}
|
|
|
|
err := it.scanItem(ctx, item)
|
|
if err != nil {
|
|
it.failErr = errs.Combine(it.failErr, err)
|
|
return false
|
|
}
|
|
|
|
it.curIndex++
|
|
it.cursor.StartStreamID = item.StreamID
|
|
it.cursor.StartPosition = item.Position
|
|
|
|
return true
|
|
}
|
|
|
|
func (it *loopSegmentIterator) doNextQuery(ctx context.Context) (_ tagsql.Rows, err error) {
|
|
defer mon.Task()(&ctx)(&err)
|
|
|
|
return it.db.db.QueryContext(ctx, `
|
|
SELECT
|
|
stream_id, position,
|
|
created_at, expires_at, repaired_at,
|
|
root_piece_id,
|
|
encrypted_size,
|
|
plain_offset, plain_size,
|
|
redundancy,
|
|
remote_alias_pieces,
|
|
placement
|
|
FROM segments
|
|
`+it.db.asOfTime(it.asOfSystemTime, it.asOfSystemInterval)+`
|
|
WHERE
|
|
(stream_id, position) > ($1, $2) AND stream_id <= $4
|
|
ORDER BY (stream_id, position) ASC
|
|
LIMIT $3
|
|
`, it.cursor.StartStreamID, it.cursor.StartPosition.Encode(),
|
|
it.batchSize, it.cursor.EndStreamID,
|
|
)
|
|
}
|
|
|
|
// scanItem scans doNextQuery results into LoopSegmentEntry.
|
|
func (it *loopSegmentIterator) scanItem(ctx context.Context, item *LoopSegmentEntry) error {
|
|
err := it.curRows.Scan(
|
|
&item.StreamID, &item.Position,
|
|
&item.CreatedAt, &item.ExpiresAt, &item.RepairedAt,
|
|
&item.RootPieceID,
|
|
&item.EncryptedSize,
|
|
&item.PlainOffset, &item.PlainSize,
|
|
redundancyScheme{&item.Redundancy},
|
|
&item.AliasPieces,
|
|
&item.Placement,
|
|
)
|
|
if err != nil {
|
|
return Error.New("failed to scan segments: %w", err)
|
|
}
|
|
|
|
// allocate new Pieces only if existing have not enough capacity
|
|
if cap(it.batchPieces[it.curIndex]) < len(item.AliasPieces) {
|
|
it.batchPieces[it.curIndex] = make(Pieces, len(item.AliasPieces))
|
|
} else {
|
|
it.batchPieces[it.curIndex] = it.batchPieces[it.curIndex][:len(item.AliasPieces)]
|
|
}
|
|
|
|
item.Pieces, err = it.db.aliasCache.convertAliasesToPieces(ctx, item.AliasPieces, it.batchPieces[it.curIndex])
|
|
if err != nil {
|
|
return Error.New("failed to convert aliases to pieces: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|