storj/pkg/audit/cursor.go

164 lines
3.9 KiB
Go
Raw Normal View History

2019-01-24 20:15:10 +00:00
// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package audit
import (
"context"
crand "crypto/rand"
"encoding/binary"
"math/rand"
"sync"
"time"
"github.com/golang/protobuf/ptypes"
"github.com/zeebo/errs"
"storj.io/storj/pkg/eestream"
"storj.io/storj/pkg/pb"
"storj.io/storj/pkg/storage/meta"
"storj.io/storj/pkg/storj"
"storj.io/storj/satellite/metainfo"
)
// Stripe keeps track of a stripe's index and its parent segment
type Stripe struct {
Index int64
Segment *pb.Pointer
2019-03-28 20:09:23 +00:00
SegmentPath storj.Path
}
// Cursor keeps track of audit location in pointer db
type Cursor struct {
metainfo *metainfo.Service
lastPath storj.Path
mutex sync.Mutex
}
// NewCursor creates a Cursor which iterates over pointer db
func NewCursor(metainfo *metainfo.Service) *Cursor {
return &Cursor{
metainfo: metainfo,
}
}
// NextStripe returns a random stripe to be audited. "more" is true except when we have completed iterating over metainfo. It can be disregarded if there is an error or stripe returned
func (cursor *Cursor) NextStripe(ctx context.Context) (stripe *Stripe, more bool, err error) {
cursor.mutex.Lock()
defer cursor.mutex.Unlock()
var pointerItems []*pb.ListResponse_Item
var path storj.Path
pointerItems, more, err = cursor.metainfo.List("", cursor.lastPath, "", true, 0, meta.None)
if err != nil {
return nil, more, err
}
// keep track of last path listed
if !more {
cursor.lastPath = ""
} else {
cursor.lastPath = pointerItems[len(pointerItems)-1].Path
}
pointer, path, err := cursor.getRandomValidPointer(pointerItems)
if err != nil {
return nil, more, err
}
if pointer == nil {
return nil, more, nil
}
index, err := getRandomStripe(pointer)
if err != nil {
return nil, more, err
}
return &Stripe{
Index: index,
Segment: pointer,
2019-03-28 20:09:23 +00:00
SegmentPath: path,
}, more, nil
}
func getRandomStripe(pointer *pb.Pointer) (index int64, err error) {
redundancy, err := eestream.NewRedundancyStrategyFromProto(pointer.GetRemote().GetRedundancy())
if err != nil {
return 0, err
}
// the last segment could be smaller than stripe size
if pointer.GetSegmentSize() < int64(redundancy.StripeSize()) {
return 0, nil
}
var src cryptoSource
rnd := rand.New(src)
numStripes := pointer.GetSegmentSize() / int64(redundancy.StripeSize())
randomStripeIndex := rnd.Int63n(numStripes)
return randomStripeIndex, nil
}
// getRandomValidPointer attempts to get a random remote pointer from a list. If it sees expired pointers in the process of looking, deletes them
func (cursor *Cursor) getRandomValidPointer(pointerItems []*pb.ListResponse_Item) (pointer *pb.Pointer, path storj.Path, err error) {
var src cryptoSource
rnd := rand.New(src)
errGroup := new(errs.Group)
randomNums := rnd.Perm(len(pointerItems))
for _, randomIndex := range randomNums {
pointerItem := pointerItems[randomIndex]
path := pointerItem.Path
// get pointer info
pointer, err := cursor.metainfo.Get(path)
if err != nil {
errGroup.Add(err)
continue
}
//delete expired items rather than auditing them
if expiration := pointer.GetExpirationDate(); expiration != nil {
t, err := ptypes.Timestamp(expiration)
if err != nil {
errGroup.Add(err)
continue
}
if t.Before(time.Now()) {
err := cursor.metainfo.Delete(path)
if err != nil {
errGroup.Add(err)
}
continue
}
}
if pointer.GetType() != pb.Pointer_REMOTE || pointer.GetSegmentSize() == 0 {
continue
}
return pointer, path, nil
}
return nil, "", errGroup.Err()
}
// cryptoSource implements the math/rand Source interface using crypto/rand
type cryptoSource struct{}
func (s cryptoSource) Seed(seed int64) {}
func (s cryptoSource) Int63() int64 {
return int64(s.Uint64() & ^uint64(1<<63))
}
func (s cryptoSource) Uint64() (v uint64) {
err := binary.Read(crand.Reader, binary.BigEndian, &v)
if err != nil {
panic(err)
}
return v
}