Use Nodes array from pdb.Get (#578)
* Use Nodes array from pdb.Get * fix problems with captplanet tests * better comments
This commit is contained in:
parent
de46a999bc
commit
deb015970d
@ -83,7 +83,7 @@ func main() {
|
||||
}
|
||||
|
||||
// Example Get
|
||||
getRes, err := client.Get(ctx, path)
|
||||
getRes, _, err := client.Get(ctx, path)
|
||||
|
||||
if err != nil {
|
||||
logger.Error("couldn't GET pointer from db", zap.Error(err))
|
||||
|
@ -72,7 +72,7 @@ func (cursor *Cursor) NextStripe(ctx context.Context) (stripe *Stripe, err error
|
||||
}
|
||||
|
||||
// get pointer info
|
||||
pointer, err := cursor.pointers.Get(ctx, path)
|
||||
pointer, _, err := cursor.pointers.Get(ctx, path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -47,7 +47,7 @@ type ListItem struct {
|
||||
// Client services offerred for the interface
|
||||
type Client interface {
|
||||
Put(ctx context.Context, path storj.Path, pointer *pb.Pointer) error
|
||||
Get(ctx context.Context, path storj.Path) (*pb.Pointer, error)
|
||||
Get(ctx context.Context, path storj.Path) (*pb.Pointer, []*pb.Node, error)
|
||||
List(ctx context.Context, prefix, startAfter, endBefore storj.Path, recursive bool, limit int, metaFlags uint32) (items []ListItem, more bool, err error)
|
||||
Delete(ctx context.Context, path storj.Path) error
|
||||
|
||||
@ -96,21 +96,21 @@ func (pdb *PointerDB) Put(ctx context.Context, path storj.Path, pointer *pb.Poin
|
||||
}
|
||||
|
||||
// Get is the interface to make a GET request, needs PATH and APIKey
|
||||
func (pdb *PointerDB) Get(ctx context.Context, path storj.Path) (pointer *pb.Pointer, err error) {
|
||||
func (pdb *PointerDB) Get(ctx context.Context, path storj.Path) (pointer *pb.Pointer, nodes []*pb.Node, err error) {
|
||||
defer mon.Task()(&ctx)(&err)
|
||||
|
||||
res, err := pdb.grpcClient.Get(ctx, &pb.GetRequest{Path: path})
|
||||
if err != nil {
|
||||
if status.Code(err) == codes.NotFound {
|
||||
return nil, storage.ErrKeyNotFound.Wrap(err)
|
||||
return nil, nil, storage.ErrKeyNotFound.Wrap(err)
|
||||
}
|
||||
return nil, Error.Wrap(err)
|
||||
return nil, nil, Error.Wrap(err)
|
||||
}
|
||||
|
||||
pdb.pba = res.GetPba()
|
||||
pdb.authorization = res.GetAuthorization()
|
||||
|
||||
return res.GetPointer(), nil
|
||||
return res.GetPointer(), res.GetNodes(), nil
|
||||
}
|
||||
|
||||
// List is the interface to make a LIST request, needs StartingPathKey, Limit, and APIKey
|
||||
|
@ -144,7 +144,7 @@ func TestGet(t *testing.T) {
|
||||
err = proto.Unmarshal(byteData, ptr)
|
||||
assert.NoError(t, err)
|
||||
|
||||
getResponse := pb.GetResponse{Pointer: ptr}
|
||||
getResponse := pb.GetResponse{Pointer: ptr, Nodes: []*pb.Node{}}
|
||||
|
||||
errTag := fmt.Sprintf("Test case #%d", i)
|
||||
|
||||
@ -153,13 +153,15 @@ func TestGet(t *testing.T) {
|
||||
|
||||
gc.EXPECT().Get(gomock.Any(), &getRequest).Return(&getResponse, tt.err)
|
||||
|
||||
pointer, err := pdb.Get(ctx, tt.path)
|
||||
pointer, nodes, err := pdb.Get(ctx, tt.path)
|
||||
|
||||
if err != nil {
|
||||
assert.True(t, strings.Contains(err.Error(), tt.errString), errTag)
|
||||
assert.Nil(t, pointer)
|
||||
assert.Nil(t, nodes)
|
||||
} else {
|
||||
assert.NotNil(t, pointer)
|
||||
assert.NotNil(t, nodes)
|
||||
assert.NoError(t, err, errTag)
|
||||
}
|
||||
}
|
||||
|
@ -48,11 +48,12 @@ func (mr *MockClientMockRecorder) Delete(arg0, arg1 interface{}) *gomock.Call {
|
||||
}
|
||||
|
||||
// Get mocks base method
|
||||
func (m *MockClient) Get(arg0 context.Context, arg1 string) (*pb.Pointer, error) {
|
||||
func (m *MockClient) Get(arg0 context.Context, arg1 string) (*pb.Pointer, []*pb.Node, error) {
|
||||
ret := m.ctrl.Call(m, "Get", arg0, arg1)
|
||||
ret0, _ := ret[0].(*pb.Pointer)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
ret1, _ := ret[1].([]*pb.Node)
|
||||
ret2, _ := ret[2].(error)
|
||||
return ret0, ret1, ret2
|
||||
}
|
||||
|
||||
// Get indicates an expected call of Get
|
||||
|
@ -13,7 +13,6 @@ import (
|
||||
"github.com/vivint/infectious"
|
||||
"go.uber.org/zap"
|
||||
monkit "gopkg.in/spacemonkeygo/monkit.v2"
|
||||
|
||||
"storj.io/storj/pkg/dht"
|
||||
"storj.io/storj/pkg/eestream"
|
||||
"storj.io/storj/pkg/node"
|
||||
@ -73,7 +72,7 @@ func NewSegmentStore(oc overlay.Client, ec ecclient.Client, pdb pdbclient.Client
|
||||
func (s *segmentStore) Meta(ctx context.Context, path storj.Path) (meta Meta, err error) {
|
||||
defer mon.Task()(&ctx)(&err)
|
||||
|
||||
pr, err := s.pdb.Get(ctx, path)
|
||||
pr, _, err := s.pdb.Get(ctx, path)
|
||||
if err != nil {
|
||||
return Meta{}, Error.Wrap(err)
|
||||
}
|
||||
@ -193,7 +192,7 @@ func (s *segmentStore) makeRemotePointer(nodes []*pb.Node, pieceID client.PieceI
|
||||
func (s *segmentStore) Get(ctx context.Context, path storj.Path) (rr ranger.Ranger, meta Meta, err error) {
|
||||
defer mon.Task()(&ctx)(&err)
|
||||
|
||||
pr, err := s.pdb.Get(ctx, path)
|
||||
pr, nodes, err := s.pdb.Get(ctx, path)
|
||||
if err != nil {
|
||||
return nil, Meta{}, Error.Wrap(err)
|
||||
}
|
||||
@ -201,10 +200,14 @@ func (s *segmentStore) Get(ctx context.Context, path storj.Path) (rr ranger.Rang
|
||||
if pr.GetType() == pb.Pointer_REMOTE {
|
||||
seg := pr.GetRemote()
|
||||
pid := client.PieceID(seg.GetPieceId())
|
||||
nodes, err := s.lookupNodes(ctx, seg)
|
||||
|
||||
// fall back if nodes are not available
|
||||
if nodes == nil {
|
||||
nodes, err = s.lookupNodes(ctx, seg)
|
||||
if err != nil {
|
||||
return nil, Meta{}, Error.Wrap(err)
|
||||
}
|
||||
}
|
||||
|
||||
es, err := makeErasureScheme(pr.GetRemote().GetRedundancy())
|
||||
if err != nil {
|
||||
@ -251,7 +254,7 @@ func makeErasureScheme(rs *pb.RedundancyScheme) (eestream.ErasureScheme, error)
|
||||
func (s *segmentStore) Delete(ctx context.Context, path storj.Path) (err error) {
|
||||
defer mon.Task()(&ctx)(&err)
|
||||
|
||||
pr, err := s.pdb.Get(ctx, path)
|
||||
pr, nodes, err := s.pdb.Get(ctx, path)
|
||||
if err != nil {
|
||||
return Error.Wrap(err)
|
||||
}
|
||||
@ -259,10 +262,14 @@ func (s *segmentStore) Delete(ctx context.Context, path storj.Path) (err error)
|
||||
if pr.GetType() == pb.Pointer_REMOTE {
|
||||
seg := pr.GetRemote()
|
||||
pid := client.PieceID(seg.PieceId)
|
||||
nodes, err := s.lookupNodes(ctx, seg)
|
||||
|
||||
// fall back if nodes are not available
|
||||
if nodes == nil {
|
||||
nodes, err = s.lookupNodes(ctx, seg)
|
||||
if err != nil {
|
||||
return Error.Wrap(err)
|
||||
}
|
||||
}
|
||||
|
||||
authorization := s.pdb.SignedMessage()
|
||||
// ecclient sends delete request
|
||||
@ -281,7 +288,7 @@ func (s *segmentStore) Repair(ctx context.Context, path storj.Path, lostPieces [
|
||||
defer mon.Task()(&ctx)(&err)
|
||||
|
||||
//Read the segment's pointer's info from the PointerDB
|
||||
pr, err := s.pdb.Get(ctx, path)
|
||||
pr, originalNodes, err := s.pdb.Get(ctx, path)
|
||||
if err != nil {
|
||||
return Error.Wrap(err)
|
||||
}
|
||||
@ -293,11 +300,14 @@ func (s *segmentStore) Repair(ctx context.Context, path storj.Path, lostPieces [
|
||||
seg := pr.GetRemote()
|
||||
pid := client.PieceID(seg.GetPieceId())
|
||||
|
||||
// fall back if nodes are not available
|
||||
if originalNodes == nil {
|
||||
// Get the list of remote pieces from the pointer
|
||||
originalNodes, err := s.lookupNodes(ctx, seg)
|
||||
originalNodes, err = s.lookupNodes(ctx, seg)
|
||||
if err != nil {
|
||||
return Error.Wrap(err)
|
||||
}
|
||||
}
|
||||
|
||||
// get the nodes list that needs to be excluded
|
||||
var excludeNodeIDs []dht.NodeID
|
||||
|
@ -12,7 +12,6 @@ import (
|
||||
"github.com/golang/mock/gomock"
|
||||
"github.com/golang/protobuf/ptypes"
|
||||
"github.com/stretchr/testify/assert"
|
||||
|
||||
"storj.io/storj/pkg/eestream"
|
||||
mock_eestream "storj.io/storj/pkg/eestream/mocks"
|
||||
mock_overlay "storj.io/storj/pkg/overlay/mocks"
|
||||
@ -72,7 +71,7 @@ func TestSegmentStoreMeta(t *testing.T) {
|
||||
calls := []*gomock.Call{
|
||||
mockPDB.EXPECT().Get(
|
||||
gomock.Any(), gomock.Any(),
|
||||
).Return(tt.returnPointer, nil),
|
||||
).Return(tt.returnPointer, nil, nil),
|
||||
}
|
||||
gomock.InOrder(calls...)
|
||||
|
||||
@ -219,7 +218,7 @@ func TestSegmentStoreGetInline(t *testing.T) {
|
||||
ExpirationDate: someTime,
|
||||
Size: tt.size,
|
||||
Metadata: tt.metadata,
|
||||
}, nil),
|
||||
}, nil, nil),
|
||||
}
|
||||
gomock.InOrder(calls...)
|
||||
|
||||
@ -282,7 +281,7 @@ func TestSegmentStoreRepairRemote(t *testing.T) {
|
||||
ExpirationDate: someTime,
|
||||
Size: tt.size,
|
||||
Metadata: tt.metadata,
|
||||
}, nil),
|
||||
}, nil, nil),
|
||||
mockOC.EXPECT().BulkLookup(gomock.Any(), gomock.Any()),
|
||||
mockOC.EXPECT().Choose(gomock.Any(), gomock.Any()).Return(tt.newNodes, nil),
|
||||
mockPDB.EXPECT().SignedMessage(),
|
||||
@ -355,7 +354,7 @@ func TestSegmentStoreGetRemote(t *testing.T) {
|
||||
ExpirationDate: someTime,
|
||||
Size: tt.size,
|
||||
Metadata: tt.metadata,
|
||||
}, nil),
|
||||
}, nil, nil),
|
||||
mockOC.EXPECT().BulkLookup(gomock.Any(), gomock.Any()),
|
||||
mockPDB.EXPECT().SignedMessage(),
|
||||
mockPDB.EXPECT().PayerBandwidthAllocation(),
|
||||
@ -409,7 +408,7 @@ func TestSegmentStoreDeleteInline(t *testing.T) {
|
||||
ExpirationDate: someTime,
|
||||
Size: tt.size,
|
||||
Metadata: tt.metadata,
|
||||
}, nil),
|
||||
}, nil, nil),
|
||||
mockPDB.EXPECT().Delete(
|
||||
gomock.Any(), gomock.Any(),
|
||||
),
|
||||
@ -469,7 +468,7 @@ func TestSegmentStoreDeleteRemote(t *testing.T) {
|
||||
ExpirationDate: someTime,
|
||||
Size: tt.size,
|
||||
Metadata: tt.metadata,
|
||||
}, nil),
|
||||
}, nil, nil),
|
||||
mockOC.EXPECT().BulkLookup(gomock.Any(), gomock.Any()),
|
||||
mockPDB.EXPECT().SignedMessage(),
|
||||
mockEC.EXPECT().Delete(
|
||||
|
Loading…
Reference in New Issue
Block a user