JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

satellite/nodeselection: use the same Node object from overlay and nodeselection

Browse Source 
We use two different Node types in `overlay` and `uploadnodeselection` and converting back and forth.

Using the same object would allow us to use a unified node selection interface everywhere.

Change-Id: Ie71e29d60184ee0e5b4547eb54325f09c418f73c
This commit is contained in:
Márton Elek 2023-06-30 12:35:07 +02:00 committed by Elek, Márton
parent 20a47034a5
commit d38b8fa2c4
27 changed files with 210 additions and 263 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
cmd/satellite/main.go
View File

@ -40,7 +40,7 @@ import (
"storj.io/storj/satellite/accounting/live"
"storj.io/storj/satellite/compensation"
"storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/overlay"
"storj.io/storj/satellite/nodeselection/uploadselection"
"storj.io/storj/satellite/payments/stripe"
"storj.io/storj/satellite/satellitedb"
)
@ -932,7 +932,7 @@ func cmdRestoreTrash(cmd *cobra.Command, args []string) error {
successes := new(int64)
failures := new(int64)
undelete := func(node *overlay.SelectedNode) {
undelete := func(node *uploadselection.SelectedNode) {
log.Info("starting restore trash", zap.String("Node ID", node.ID.String()))
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
@ -966,9 +966,9 @@ func cmdRestoreTrash(cmd *cobra.Command, args []string) error {
log.Info("successful restore trash", zap.String("Node ID", node.ID.String()))
}
var nodes []*overlay.SelectedNode
var nodes []*uploadselection.SelectedNode
if len(args) == 0 {
err = db.OverlayCache().IterateAllContactedNodes(ctx, func(ctx context.Context, node *overlay.SelectedNode) error {
err = db.OverlayCache().IterateAllContactedNodes(ctx, func(ctx context.Context, node *uploadselection.SelectedNode) error {
nodes = append(nodes, node)
return nil
})
@ -985,7 +985,7 @@ func cmdRestoreTrash(cmd *cobra.Command, args []string) error {
if err != nil {
return err
}
nodes = append(nodes, &overlay.SelectedNode{
nodes = append(nodes, &uploadselection.SelectedNode{
ID: dossier.Id,
Address: dossier.Address,
LastNet: dossier.LastNet,

3
cmd/tools/segment-verify/node_check.go
View File

@ -15,6 +15,7 @@ import (
"storj.io/common/uuid"
"storj.io/private/process"
"storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/nodeselection/uploadselection"
"storj.io/storj/satellite/overlay"
"storj.io/storj/satellite/satellitedb"
)
@ -78,7 +79,7 @@ type NodeCheckConfig struct {
// NodeCheckOverlayDB contains dependencies from overlay that are needed for the processing.
type NodeCheckOverlayDB interface {
IterateAllContactedNodes(context.Context, func(context.Context, *overlay.SelectedNode) error) error
IterateAllContactedNodes(context.Context, func(context.Context, *uploadselection.SelectedNode) error) error
IterateAllNodeDossiers(context.Context, func(context.Context, *overlay.NodeDossier) error) error
}

3
cmd/tools/segment-verify/service.go
View File

@ -21,6 +21,7 @@ import (
"storj.io/common/uuid"
"storj.io/storj/satellite/audit"
"storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/nodeselection/uploadselection"
"storj.io/storj/satellite/overlay"
)
@ -46,7 +47,7 @@ type Verifier interface {
type Overlay interface {
// Get looks up the node by nodeID
Get(ctx context.Context, nodeID storj.NodeID) (*overlay.NodeDossier, error)
SelectAllStorageNodesDownload(ctx context.Context, onlineWindow time.Duration, asOf overlay.AsOfSystemTimeConfig) ([]*overlay.SelectedNode, error)
SelectAllStorageNodesDownload(ctx context.Context, onlineWindow time.Duration, asOf overlay.AsOfSystemTimeConfig) ([]*uploadselection.SelectedNode, error)
}
// SegmentWriter allows writing segments to some output.

7
cmd/tools/segment-verify/service_test.go
View File

@ -23,6 +23,7 @@ import (
segmentverify "storj.io/storj/cmd/tools/segment-verify"
"storj.io/storj/private/testplanet"
"storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/nodeselection/uploadselection"
"storj.io/storj/satellite/overlay"
)
@ -344,10 +345,10 @@ func (db *metabaseMock) Get(ctx context.Context, nodeID storj.NodeID) (*overlay.
}, nil
}
func (db *metabaseMock) SelectAllStorageNodesDownload(ctx context.Context, onlineWindow time.Duration, asOf overlay.AsOfSystemTimeConfig) ([]*overlay.SelectedNode, error) {
var xs []*overlay.SelectedNode
func (db *metabaseMock) SelectAllStorageNodesDownload(ctx context.Context, onlineWindow time.Duration, asOf overlay.AsOfSystemTimeConfig) ([]*uploadselection.SelectedNode, error) {
var xs []*uploadselection.SelectedNode
for nodeID := range db.nodeIDToAlias {
xs = append(xs, &overlay.SelectedNode{
xs = append(xs, &uploadselection.SelectedNode{
ID: nodeID,
Address: &pb.NodeAddress{
Address: fmt.Sprintf("nodeid:%v", nodeID),

2
satellite/nodeselection/uploadselection/criteria.go
View File

@ -17,7 +17,7 @@ type Criteria struct {
}
// MatchInclude returns with true if node is selected.
func (c *Criteria) MatchInclude(node *Node) bool {
func (c *Criteria) MatchInclude(node *SelectedNode) bool {
if ContainsID(c.ExcludeNodeIDs, node.ID) {
return false
}

38
satellite/nodeselection/uploadselection/criteria_test.go
View File

@ -19,15 +19,15 @@ func TestCriteria_AutoExcludeSubnet(t *testing.T) {
AutoExcludeSubnets: map[string]struct{}{},
}
assert.True(t, criteria.MatchInclude(&Node{
assert.True(t, criteria.MatchInclude(&SelectedNode{
LastNet: "192.168.0.1",
}))
assert.False(t, criteria.MatchInclude(&Node{
assert.False(t, criteria.MatchInclude(&SelectedNode{
LastNet: "192.168.0.1",
}))
assert.True(t, criteria.MatchInclude(&Node{
assert.True(t, criteria.MatchInclude(&SelectedNode{
LastNet: "192.168.1.1",
}))
}
@ -40,18 +40,12 @@ func TestCriteria_ExcludeNodeID(t *testing.T) {
ExcludeNodeIDs: []storj.NodeID{excluded},
}
assert.False(t, criteria.MatchInclude(&Node{
NodeURL: storj.NodeURL{
ID: excluded,
Address: "localhost",
},
assert.False(t, criteria.MatchInclude(&SelectedNode{
ID: excluded,
}))
assert.True(t, criteria.MatchInclude(&Node{
NodeURL: storj.NodeURL{
ID: included,
Address: "localhost",
},
assert.True(t, criteria.MatchInclude(&SelectedNode{
ID: included,
}))
}
@ -65,20 +59,16 @@ func TestCriteria_NodeIDAndSubnet(t *testing.T) {
}
// due to node id criteria
assert.False(t, criteria.MatchInclude(&Node{
NodeURL: storj.NodeURL{
ID: excluded,
Address: "192.168.0.1",
},
assert.False(t, criteria.MatchInclude(&SelectedNode{
ID: excluded,
LastNet: "192.168.0.1",
}))
// should be included as previous one excluded and
// not stored for subnet exclusion
assert.True(t, criteria.MatchInclude(&Node{
NodeURL: storj.NodeURL{
ID: testrand.NodeID(),
Address: "192.168.0.2",
},
assert.True(t, criteria.MatchInclude(&SelectedNode{
ID: testrand.NodeID(),
LastNet: "192.168.0.2",
}))
}
@ -132,7 +122,7 @@ func TestCriteria_Geofencing(t *testing.T) {
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
assert.Equal(t, c.expected, c.criteria.MatchInclude(&Node{
assert.Equal(t, c.expected, c.criteria.MatchInclude(&SelectedNode{
CountryCode: c.country,
}))
})

42
satellite/nodeselection/uploadselection/node.go
View File

@ -4,22 +4,52 @@
package uploadselection
import (
"time"
"github.com/zeebo/errs"
"storj.io/common/pb"
"storj.io/common/storj"
"storj.io/common/storj/location"
)
// Node defines necessary information for node-selection.
type Node struct {
storj.NodeURL
// NodeTag is a tag associated with a node (approved by signer).
type NodeTag struct {
NodeID storj.NodeID
Timestamp time.Time
Signer storj.NodeID
Name string
Value []byte
}
// NodeTags is a collection of multiple NodeTag.
type NodeTags []NodeTag
// FindBySignerAndName selects first tag with same name / NodeID.
func (n NodeTags) FindBySignerAndName(signer storj.NodeID, name string) (NodeTag, error) {
for _, tag := range n {
if tag.Name == name && signer == tag.Signer {
return tag, nil
}
}
return NodeTag{}, errs.New("tags not found")
}
// SelectedNode is used as a result for creating orders limits.
type SelectedNode struct {
ID storj.NodeID
Address *pb.NodeAddress
LastNet string
LastIPPort string
CountryCode location.CountryCode
}
// Clone returns a deep clone of the selected node.
func (node *Node) Clone() *Node {
return &Node{
NodeURL: node.NodeURL,
func (node *SelectedNode) Clone() *SelectedNode {
copy := pb.CopyNode(&pb.Node{Id: node.ID, Address: node.Address})
return &SelectedNode{
ID: copy.Id,
Address: copy.Address,
LastNet: node.LastNet,
LastIPPort: node.LastIPPort,
CountryCode: node.CountryCode,

16
satellite/nodeselection/uploadselection/selector.go
View File

@ -8,7 +8,7 @@ import (
)
// SelectByID implements selection from nodes with every node having equal probability.
type SelectByID []*Node
type SelectByID []*SelectedNode
var _ Selector = (SelectByID)(nil)
@ -16,12 +16,12 @@ var _ Selector = (SelectByID)(nil)
func (nodes SelectByID) Count() int { return len(nodes) }
// Select selects upto n nodes.
func (nodes SelectByID) Select(n int, criteria Criteria) []*Node {
func (nodes SelectByID) Select(n int, criteria Criteria) []*SelectedNode {
if n <= 0 {
return nil
}
selected := []*Node{}
selected := []*SelectedNode{}
for _, idx := range mathrand.Perm(len(nodes)) {
node := nodes[idx]
@ -46,12 +46,12 @@ var _ Selector = (SelectBySubnet)(nil)
// Subnet groups together nodes with the same subnet.
type Subnet struct {
Net string
Nodes []*Node
Nodes []*SelectedNode
}
// SelectBySubnetFromNodes creates SelectBySubnet selector from nodes.
func SelectBySubnetFromNodes(nodes []*Node) SelectBySubnet {
bynet := map[string][]*Node{}
func SelectBySubnetFromNodes(nodes []*SelectedNode) SelectBySubnet {
bynet := map[string][]*SelectedNode{}
for _, node := range nodes {
bynet[node.LastNet] = append(bynet[node.LastNet], node)
}
@ -71,12 +71,12 @@ func SelectBySubnetFromNodes(nodes []*Node) SelectBySubnet {
func (subnets SelectBySubnet) Count() int { return len(subnets) }
// Select selects upto n nodes.
func (subnets SelectBySubnet) Select(n int, criteria Criteria) []*Node {
func (subnets SelectBySubnet) Select(n int, criteria Criteria) []*SelectedNode {
if n <= 0 {
return nil
}
selected := []*Node{}
selected := []*SelectedNode{}
for _, idx := range mathrand.Perm(len(subnets)) {
subnet := subnets[idx]
node := subnet.Nodes[mathrand.Intn(len(subnet.Nodes))]

92
satellite/nodeselection/uploadselection/selector_test.go
View File

@ -24,34 +24,25 @@ func TestSelectByID(t *testing.T) {
// create 3 nodes, 2 with same subnet
lastNetDuplicate := "1.0.1"
subnetA1 := &uploadselection.Node{
NodeURL: storj.NodeURL{
ID: testrand.NodeID(),
Address: lastNetDuplicate + ".4:8080",
},
subnetA1 := &uploadselection.SelectedNode{
ID: testrand.NodeID(),
LastNet: lastNetDuplicate,
LastIPPort: lastNetDuplicate + ".4:8080",
}
subnetA2 := &uploadselection.Node{
NodeURL: storj.NodeURL{
ID: testrand.NodeID(),
Address: lastNetDuplicate + ".5:8080",
},
subnetA2 := &uploadselection.SelectedNode{
ID: testrand.NodeID(),
LastNet: lastNetDuplicate,
LastIPPort: lastNetDuplicate + ".5:8080",
}
lastNetSingle := "1.0.2"
subnetB1 := &uploadselection.Node{
NodeURL: storj.NodeURL{
ID: testrand.NodeID(),
Address: lastNetSingle + ".5:8080",
},
subnetB1 := &uploadselection.SelectedNode{
ID: testrand.NodeID(),
LastNet: lastNetSingle,
LastIPPort: lastNetSingle + ".5:8080",
}
nodes := []*uploadselection.Node{subnetA1, subnetA2, subnetB1}
nodes := []*uploadselection.SelectedNode{subnetA1, subnetA2, subnetB1}
selector := uploadselection.SelectByID(nodes)
const (
@ -93,34 +84,25 @@ func TestSelectBySubnet(t *testing.T) {
// create 3 nodes, 2 with same subnet
lastNetDuplicate := "1.0.1"
subnetA1 := &uploadselection.Node{
NodeURL: storj.NodeURL{
ID: testrand.NodeID(),
Address: lastNetDuplicate + ".4:8080",
},
subnetA1 := &uploadselection.SelectedNode{
ID: testrand.NodeID(),
LastNet: lastNetDuplicate,
LastIPPort: lastNetDuplicate + ".4:8080",
}
subnetA2 := &uploadselection.Node{
NodeURL: storj.NodeURL{
ID: testrand.NodeID(),
Address: lastNetDuplicate + ".5:8080",
},
subnetA2 := &uploadselection.SelectedNode{
ID: testrand.NodeID(),
LastNet: lastNetDuplicate,
LastIPPort: lastNetDuplicate + ".5:8080",
}
lastNetSingle := "1.0.2"
subnetB1 := &uploadselection.Node{
NodeURL: storj.NodeURL{
ID: testrand.NodeID(),
Address: lastNetSingle + ".5:8080",
},
subnetB1 := &uploadselection.SelectedNode{
ID: testrand.NodeID(),
LastNet: lastNetSingle,
LastIPPort: lastNetSingle + ".5:8080",
}
nodes := []*uploadselection.Node{subnetA1, subnetA2, subnetB1}
nodes := []*uploadselection.SelectedNode{subnetA1, subnetA2, subnetB1}
selector := uploadselection.SelectBySubnetFromNodes(nodes)
const (
@ -174,34 +156,25 @@ func TestSelectBySubnetOneAtATime(t *testing.T) {
// create 3 nodes, 2 with same subnet
lastNetDuplicate := "1.0.1"
subnetA1 := &uploadselection.Node{
NodeURL: storj.NodeURL{
ID: testrand.NodeID(),
Address: lastNetDuplicate + ".4:8080",
},
subnetA1 := &uploadselection.SelectedNode{
ID: testrand.NodeID(),
LastNet: lastNetDuplicate,
LastIPPort: lastNetDuplicate + ".4:8080",
}
subnetA2 := &uploadselection.Node{
NodeURL: storj.NodeURL{
ID: testrand.NodeID(),
Address: lastNetDuplicate + ".5:8080",
},
subnetA2 := &uploadselection.SelectedNode{
ID: testrand.NodeID(),
LastNet: lastNetDuplicate,
LastIPPort: lastNetDuplicate + ".5:8080",
}
lastNetSingle := "1.0.2"
subnetB1 := &uploadselection.Node{
NodeURL: storj.NodeURL{
ID: testrand.NodeID(),
Address: lastNetSingle + ".5:8080",
},
subnetB1 := &uploadselection.SelectedNode{
ID: testrand.NodeID(),
LastNet: lastNetSingle,
LastIPPort: lastNetSingle + ".5:8080",
}
nodes := []*uploadselection.Node{subnetA1, subnetA2, subnetB1}
nodes := []*uploadselection.SelectedNode{subnetA1, subnetA2, subnetB1}
selector := uploadselection.SelectBySubnetFromNodes(nodes)
const (
@ -247,37 +220,28 @@ func TestSelectFiltered(t *testing.T) {
// create 3 nodes, 2 with same subnet
lastNetDuplicate := "1.0.1"
firstID := testrand.NodeID()
subnetA1 := &uploadselection.Node{
NodeURL: storj.NodeURL{
ID: firstID,
Address: lastNetDuplicate + ".4:8080",
},
subnetA1 := &uploadselection.SelectedNode{
ID: firstID,
LastNet: lastNetDuplicate,
LastIPPort: lastNetDuplicate + ".4:8080",
}
secondID := testrand.NodeID()
subnetA2 := &uploadselection.Node{
NodeURL: storj.NodeURL{
ID: secondID,
Address: lastNetDuplicate + ".5:8080",
},
subnetA2 := &uploadselection.SelectedNode{
ID: secondID,
LastNet: lastNetDuplicate,
LastIPPort: lastNetDuplicate + ".5:8080",
}
thirdID := testrand.NodeID()
lastNetSingle := "1.0.2"
subnetB1 := &uploadselection.Node{
NodeURL: storj.NodeURL{
ID: thirdID,
Address: lastNetSingle + ".5:8080",
},
subnetB1 := &uploadselection.SelectedNode{
ID: thirdID,
LastNet: lastNetSingle,
LastIPPort: lastNetSingle + ".5:8080",
}
nodes := []*uploadselection.Node{subnetA1, subnetA2, subnetB1}
nodes := []*uploadselection.SelectedNode{subnetA1, subnetA2, subnetB1}
selector := uploadselection.SelectByID(nodes)
assert.Len(t, selector.Select(3, uploadselection.Criteria{}), 3)

8
satellite/nodeselection/uploadselection/state.go
View File

@ -42,11 +42,11 @@ type Selector interface {
Count() int
// Select selects up-to n nodes which are included by the criteria.
// empty criteria includes all the nodes
Select(n int, criteria Criteria) []*Node
Select(n int, criteria Criteria) []*SelectedNode
}
// NewState returns a state based on the input.
func NewState(reputableNodes, newNodes []*Node) *State {
func NewState(reputableNodes, newNodes []*SelectedNode) *State {
state := &State{}
state.netByID = map[storj.NodeID]string{}
@ -78,7 +78,7 @@ type Request struct {
}
// Select selects requestedCount nodes where there will be newFraction nodes.
func (state *State) Select(ctx context.Context, request Request) (_ []*Node, err error) {
func (state *State) Select(ctx context.Context, request Request) (_ []*SelectedNode, err error) {
defer mon.Task()(&ctx)(&err)
state.mu.RLock()
@ -87,7 +87,7 @@ func (state *State) Select(ctx context.Context, request Request) (_ []*Node, err
totalCount := request.Count
newCount := int(float64(totalCount) * request.NewFraction)
var selected []*Node
var selected []*SelectedNode
var reputableNodes Selector
var newNodes Selector

21
satellite/nodeselection/uploadselection/state_test.go
View File

@ -10,7 +10,6 @@ import (
"github.com/stretchr/testify/require"
"golang.org/x/sync/errgroup"
"storj.io/common/storj"
"storj.io/common/testcontext"
"storj.io/common/testrand"
"storj.io/storj/satellite/nodeselection/uploadselection"
@ -172,15 +171,13 @@ func TestState_Select_Concurrent(t *testing.T) {
}
// createRandomNodes creates n random nodes all in the subnet.
func createRandomNodes(n int, subnet string, shareNets bool) []*uploadselection.Node {
xs := make([]*uploadselection.Node, n)
func createRandomNodes(n int, subnet string, shareNets bool) []*uploadselection.SelectedNode {
xs := make([]*uploadselection.SelectedNode, n)
for i := range xs {
addr := subnet + "." + strconv.Itoa(i) + ":8080"
xs[i] = &uploadselection.Node{
NodeURL: storj.NodeURL{
ID: testrand.NodeID(),
Address: addr,
},
xs[i] = &uploadselection.SelectedNode{
ID: testrand.NodeID(),
LastNet: addr,
LastIPPort: addr,
}
if shareNets {
@ -193,8 +190,8 @@ func createRandomNodes(n int, subnet string, shareNets bool) []*uploadselection.
}
// joinNodes appends all slices into a single slice.
func joinNodes(lists ...[]*uploadselection.Node) []*uploadselection.Node {
xs := []*uploadselection.Node{}
func joinNodes(lists ...[]*uploadselection.SelectedNode) []*uploadselection.SelectedNode {
xs := []*uploadselection.SelectedNode{}
for _, list := range lists {
xs = append(xs, list...)
}
@ -202,8 +199,8 @@ func joinNodes(lists ...[]*uploadselection.Node) []*uploadselection.Node {
}
// intersectLists returns nodes that exist in both lists compared by ID.
func intersectLists(as, bs []*uploadselection.Node) []*uploadselection.Node {
var xs []*uploadselection.Node
func intersectLists(as, bs []*uploadselection.SelectedNode) []*uploadselection.SelectedNode {
var xs []*uploadselection.SelectedNode
next:
for _, a := range as {

5
satellite/orders/mock_test.go
View File

@ -11,6 +11,7 @@ import (
gomock "github.com/golang/mock/gomock"
storj "storj.io/common/storj"
"storj.io/storj/satellite/nodeselection/uploadselection"
overlay "storj.io/storj/satellite/overlay"
)
@ -38,10 +39,10 @@ func (m *MockOverlayForOrders) EXPECT() *MockOverlayForOrdersMockRecorder {
}
// CachedGetOnlineNodesForGet mocks base method.
func (m *MockOverlayForOrders) CachedGetOnlineNodesForGet(arg0 context.Context, arg1 []storj.NodeID) (map[storj.NodeID]*overlay.SelectedNode, error) {
func (m *MockOverlayForOrders) CachedGetOnlineNodesForGet(arg0 context.Context, arg1 []storj.NodeID) (map[storj.NodeID]*uploadselection.SelectedNode, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "CachedGetOnlineNodesForGet", arg0, arg1)
ret0, _ := ret[0].(map[storj.NodeID]*overlay.SelectedNode)
ret0, _ := ret[0].(map[storj.NodeID]*uploadselection.SelectedNode)
ret1, _ := ret[1].(error)
return ret0, ret1
}

11
satellite/orders/service.go
View File

@ -18,6 +18,7 @@ import (
"storj.io/common/storj"
"storj.io/storj/satellite/internalpb"
"storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/nodeselection/uploadselection"
"storj.io/storj/satellite/overlay"
)
@ -43,7 +44,7 @@ type Config struct {
//
//go:generate mockgen -destination mock_test.go -package orders . OverlayForOrders
type Overlay interface {
CachedGetOnlineNodesForGet(context.Context, []storj.NodeID) (map[storj.NodeID]*overlay.SelectedNode, error)
CachedGetOnlineNodesForGet(context.Context, []storj.NodeID) (map[storj.NodeID]*uploadselection.SelectedNode, error)
GetOnlineNodesForAuditRepair(context.Context, []storj.NodeID) (map[storj.NodeID]*overlay.NodeReputation, error)
Get(ctx context.Context, nodeID storj.NodeID) (*overlay.NodeDossier, error)
IsOnline(node *overlay.NodeDossier) bool
@ -235,7 +236,7 @@ func getLimitByStorageNodeID(limits []*pb.AddressedOrderLimit, storageNodeID sto
}
// CreatePutOrderLimits creates the order limits for uploading pieces to nodes.
func (service *Service) CreatePutOrderLimits(ctx context.Context, bucket metabase.BucketLocation, nodes []*overlay.SelectedNode, pieceExpiration time.Time, maxPieceSize int64) (_ storj.PieceID, _ []*pb.AddressedOrderLimit, privateKey storj.PiecePrivateKey, err error) {
func (service *Service) CreatePutOrderLimits(ctx context.Context, bucket metabase.BucketLocation, nodes []*uploadselection.SelectedNode, pieceExpiration time.Time, maxPieceSize int64) (_ storj.PieceID, _ []*pb.AddressedOrderLimit, privateKey storj.PiecePrivateKey, err error) {
defer mon.Task()(&ctx)(&err)
signer, err := NewSignerPut(service, pieceExpiration, time.Now(), maxPieceSize, bucket)
@ -254,7 +255,7 @@ func (service *Service) CreatePutOrderLimits(ctx context.Context, bucket metabas
}
// ReplacePutOrderLimits replaces order limits for uploading pieces to nodes.
func (service *Service) ReplacePutOrderLimits(ctx context.Context, rootPieceID storj.PieceID, addressedLimits []*pb.AddressedOrderLimit, nodes []*overlay.SelectedNode, pieceNumbers []int32) (_ []*pb.AddressedOrderLimit, err error) {
func (service *Service) ReplacePutOrderLimits(ctx context.Context, rootPieceID storj.PieceID, addressedLimits []*pb.AddressedOrderLimit, nodes []*uploadselection.SelectedNode, pieceNumbers []int32) (_ []*pb.AddressedOrderLimit, err error) {
defer mon.Task()(&ctx)(&err)
pieceIDDeriver := rootPieceID.Deriver()
@ -457,7 +458,7 @@ func (service *Service) CreateGetRepairOrderLimits(ctx context.Context, segment
}
// CreatePutRepairOrderLimits creates the order limits for uploading the repaired pieces of segment to newNodes.
func (service *Service) CreatePutRepairOrderLimits(ctx context.Context, segment metabase.Segment, getOrderLimits []*pb.AddressedOrderLimit, healthySet map[int32]struct{}, newNodes []*overlay.SelectedNode, optimalThresholdMultiplier float64, numPiecesInExcludedCountries int) (_ []*pb.AddressedOrderLimit, _ storj.PiecePrivateKey, err error) {
func (service *Service) CreatePutRepairOrderLimits(ctx context.Context, segment metabase.Segment, getOrderLimits []*pb.AddressedOrderLimit, healthySet map[int32]struct{}, newNodes []*uploadselection.SelectedNode, optimalThresholdMultiplier float64, numPiecesInExcludedCountries int) (_ []*pb.AddressedOrderLimit, _ storj.PiecePrivateKey, err error) {
defer mon.Task()(&ctx)(&err)
// Create the order limits for being used to upload the repaired pieces
@ -590,7 +591,7 @@ func (service *Service) DecryptOrderMetadata(ctx context.Context, order *pb.Orde
return key.DecryptMetadata(order.SerialNumber, order.EncryptedMetadata)
}
func resolveStorageNode_Selected(node *overlay.SelectedNode, resolveDNS bool) *pb.Node {
func resolveStorageNode_Selected(node *uploadselection.SelectedNode, resolveDNS bool) *pb.Node {
return resolveStorageNode(&pb.Node{
Id: node.ID,
Address: node.Address,

6
satellite/orders/service_test.go
View File

@ -19,8 +19,8 @@ import (
"storj.io/common/testcontext"
"storj.io/common/testrand"
"storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/nodeselection/uploadselection"
"storj.io/storj/satellite/orders"
"storj.io/storj/satellite/overlay"
)
func TestGetOrderLimits(t *testing.T) {
@ -30,10 +30,10 @@ func TestGetOrderLimits(t *testing.T) {
bucket := metabase.BucketLocation{ProjectID: testrand.UUID(), BucketName: "bucket1"}
pieces := metabase.Pieces{}
nodes := map[storj.NodeID]*overlay.SelectedNode{}
nodes := map[storj.NodeID]*uploadselection.SelectedNode{}
for i := 0; i < 8; i++ {
nodeID := testrand.NodeID()
nodes[nodeID] = &overlay.SelectedNode{
nodes[nodeID] = &uploadselection.SelectedNode{
ID: nodeID,
Address: &pb.NodeAddress{
Address: fmt.Sprintf("host%d.com", i),

15
satellite/overlay/downloadselection.go
View File

@ -11,6 +11,7 @@ import (
"storj.io/common/storj"
"storj.io/common/sync2"
"storj.io/storj/satellite/nodeselection/uploadselection"
)
// DownloadSelectionDB implements the database for download selection cache.
@ -18,7 +19,7 @@ import (
// architecture: Database
type DownloadSelectionDB interface {
// SelectAllStorageNodesDownload returns nodes that are ready for downloading
SelectAllStorageNodesDownload(ctx context.Context, onlineWindow time.Duration, asOf AsOfSystemTimeConfig) ([]*SelectedNode, error)
SelectAllStorageNodesDownload(ctx context.Context, onlineWindow time.Duration, asOf AsOfSystemTimeConfig) ([]*uploadselection.SelectedNode, error)
}
// DownloadSelectionCacheConfig contains configuration for the selection cache.
@ -88,7 +89,7 @@ func (cache *DownloadSelectionCache) GetNodeIPsFromPlacement(ctx context.Context
}
// GetNodes gets nodes by ID from the cache, and refreshes the cache if it is stale.
func (cache *DownloadSelectionCache) GetNodes(ctx context.Context, nodes []storj.NodeID) (_ map[storj.NodeID]*SelectedNode, err error) {
func (cache *DownloadSelectionCache) GetNodes(ctx context.Context, nodes []storj.NodeID) (_ map[storj.NodeID]*uploadselection.SelectedNode, err error) {
defer mon.Task()(&ctx)(&err)
state, err := cache.cache.Get(ctx, time.Now())
@ -110,12 +111,12 @@ func (cache *DownloadSelectionCache) Size(ctx context.Context) (int, error) {
// DownloadSelectionCacheState contains state of download selection cache.
type DownloadSelectionCacheState struct {
// byID returns IP based on storj.NodeID
byID map[storj.NodeID]*SelectedNode // TODO: optimize, avoid pointery structures for performance
byID map[storj.NodeID]*uploadselection.SelectedNode // TODO: optimize, avoid pointery structures for performance
}
// NewDownloadSelectionCacheState creates a new state from the nodes.
func NewDownloadSelectionCacheState(nodes []*SelectedNode) *DownloadSelectionCacheState {
byID := map[storj.NodeID]*SelectedNode{}
func NewDownloadSelectionCacheState(nodes []*uploadselection.SelectedNode) *DownloadSelectionCacheState {
byID := map[storj.NodeID]*uploadselection.SelectedNode{}
for _, n := range nodes {
byID[n.ID] = n
}
@ -152,8 +153,8 @@ func (state *DownloadSelectionCacheState) IPsFromPlacement(nodes []storj.NodeID,
}
// Nodes returns node ip:port for nodes that are in state.
func (state *DownloadSelectionCacheState) Nodes(nodes []storj.NodeID) map[storj.NodeID]*SelectedNode {
xs := make(map[storj.NodeID]*SelectedNode, len(nodes))
func (state *DownloadSelectionCacheState) Nodes(nodes []storj.NodeID) map[storj.NodeID]*uploadselection.SelectedNode {
xs := make(map[storj.NodeID]*uploadselection.SelectedNode, len(nodes))
for _, nodeID := range nodes {
if n, exists := state.byID[nodeID]; exists {
xs[nodeID] = n.Clone() // TODO: optimize the clones

5
satellite/overlay/downloadselection_test.go
View File

@ -16,6 +16,7 @@ import (
"storj.io/common/testcontext"
"storj.io/common/testrand"
"storj.io/storj/satellite"
"storj.io/storj/satellite/nodeselection/uploadselection"
"storj.io/storj/satellite/overlay"
"storj.io/storj/satellite/satellitedb/satellitedbtest"
)
@ -87,7 +88,7 @@ func TestDownloadSelectionCacheState_IPs(t *testing.T) {
ctx := testcontext.New(t)
defer ctx.Cleanup()
node := &overlay.SelectedNode{
node := &uploadselection.SelectedNode{
ID: testrand.NodeID(),
Address: &pb.NodeAddress{
Address: "1.0.1.1:8080",
@ -96,7 +97,7 @@ func TestDownloadSelectionCacheState_IPs(t *testing.T) {
LastIPPort: "1.0.1.1:8080",
}
state := overlay.NewDownloadSelectionCacheState([]*overlay.SelectedNode{node})
state := overlay.NewDownloadSelectionCacheState([]*uploadselection.SelectedNode{node})
require.Equal(t, state.Size(), 1)
ips := state.IPs([]storj.NodeID{testrand.NodeID(), node.ID})

7
satellite/overlay/selection_test.go
View File

@ -26,6 +26,7 @@ import (
"storj.io/common/testcontext"
"storj.io/storj/private/testplanet"
"storj.io/storj/satellite"
"storj.io/storj/satellite/nodeselection/uploadselection"
"storj.io/storj/satellite/overlay"
"storj.io/storj/satellite/reputation"
)
@ -147,10 +148,10 @@ func TestOnlineOffline(t *testing.T) {
require.Empty(t, offline)
require.Len(t, online, 2)
require.False(t, slices.ContainsFunc(online, func(node overlay.SelectedNode) bool {
require.False(t, slices.ContainsFunc(online, func(node uploadselection.SelectedNode) bool {
return node.ID == unreliableNodeID
}))
require.False(t, slices.ContainsFunc(offline, func(node overlay.SelectedNode) bool {
require.False(t, slices.ContainsFunc(offline, func(node uploadselection.SelectedNode) bool {
return node.ID == unreliableNodeID
}))
})
@ -192,7 +193,7 @@ func TestEnsureMinimumRequested(t *testing.T) {
reputable := map[storj.NodeID]bool{}
countReputable := func(selected []*overlay.SelectedNode) (count int) {
countReputable := func(selected []*uploadselection.SelectedNode) (count int) {
for _, n := range selected {
if reputable[n.ID] {
count++

44
satellite/overlay/service.go
View File

@ -21,6 +21,7 @@ import (
"storj.io/storj/satellite/geoip"
"storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/nodeevents"
"storj.io/storj/satellite/nodeselection/uploadselection"
)
// ErrEmptyNode is returned when the nodeID is empty.
@ -53,18 +54,18 @@ type DB interface {
// current reputation status.
GetOnlineNodesForAuditRepair(ctx context.Context, nodeIDs []storj.NodeID, onlineWindow time.Duration) (map[storj.NodeID]*NodeReputation, error)
// SelectStorageNodes looks up nodes based on criteria
SelectStorageNodes(ctx context.Context, totalNeededNodes, newNodeCount int, criteria *NodeCriteria) ([]*SelectedNode, error)
SelectStorageNodes(ctx context.Context, totalNeededNodes, newNodeCount int, criteria *NodeCriteria) ([]*uploadselection.SelectedNode, error)
// SelectAllStorageNodesUpload returns all nodes that qualify to store data, organized as reputable nodes and new nodes
SelectAllStorageNodesUpload(ctx context.Context, selectionCfg NodeSelectionConfig) (reputable, new []*SelectedNode, err error)
SelectAllStorageNodesUpload(ctx context.Context, selectionCfg NodeSelectionConfig) (reputable, new []*uploadselection.SelectedNode, err error)
// SelectAllStorageNodesDownload returns a nodes that are ready for downloading
SelectAllStorageNodesDownload(ctx context.Context, onlineWindow time.Duration, asOf AsOfSystemTimeConfig) ([]*SelectedNode, error)
SelectAllStorageNodesDownload(ctx context.Context, onlineWindow time.Duration, asOf AsOfSystemTimeConfig) ([]*uploadselection.SelectedNode, error)
// Get looks up the node by nodeID
Get(ctx context.Context, nodeID storj.NodeID) (*NodeDossier, error)
// KnownReliableInExcludedCountries filters healthy nodes that are in excluded countries.
KnownReliableInExcludedCountries(context.Context, *NodeCriteria, storj.NodeIDList) (storj.NodeIDList, error)
// KnownReliable filters a set of nodes to reliable (online and qualified) nodes.
KnownReliable(ctx context.Context, nodeIDs storj.NodeIDList, onlineWindow, asOfSystemInterval time.Duration) (online []SelectedNode, offline []SelectedNode, err error)
KnownReliable(ctx context.Context, nodeIDs storj.NodeIDList, onlineWindow, asOfSystemInterval time.Duration) (online []uploadselection.SelectedNode, offline []uploadselection.SelectedNode, err error)
// Reliable returns all nodes that are reliable
Reliable(context.Context, *NodeCriteria) (storj.NodeIDList, error)
// UpdateReputation updates the DB columns for all reputation fields in ReputationStatus.
@ -131,7 +132,7 @@ type DB interface {
OneTimeFixLastNets(ctx context.Context) error
// IterateAllContactedNodes will call cb on all known nodes (used in restore trash contexts).
IterateAllContactedNodes(context.Context, func(context.Context, *SelectedNode) error) error
IterateAllContactedNodes(context.Context, func(context.Context, *uploadselection.SelectedNode) error) error
// IterateAllNodeDossiers will call cb on all known nodes (used for invoice generation).
IterateAllNodeDossiers(context.Context, func(context.Context, *NodeDossier) error) error
}
@ -273,15 +274,6 @@ type NodeLastContact struct {
LastContactFailure time.Time
}
// SelectedNode is used as a result for creating orders limits.
type SelectedNode struct {
ID storj.NodeID
Address *pb.NodeAddress
LastNet string
LastIPPort string
CountryCode location.CountryCode
}
// NodeReputation is used as a result for creating orders limits for audits.
type NodeReputation struct {
ID storj.NodeID
@ -291,18 +283,6 @@ type NodeReputation struct {
Reputation ReputationStatus
}
// Clone returns a deep clone of the selected node.
func (node *SelectedNode) Clone() *SelectedNode {
copy := pb.CopyNode(&pb.Node{Id: node.ID, Address: node.Address})
return &SelectedNode{
ID: copy.Id,
Address: copy.Address,
LastNet: node.LastNet,
LastIPPort: node.LastIPPort,
CountryCode: node.CountryCode,
}
}
// Service is used to store and handle node information.
//
// architecture: Service
@ -392,7 +372,7 @@ func (service *Service) Get(ctx context.Context, nodeID storj.NodeID) (_ *NodeDo
}
// CachedGetOnlineNodesForGet returns a map of nodes from the download selection cache from the suppliedIDs.
func (service *Service) CachedGetOnlineNodesForGet(ctx context.Context, nodeIDs []storj.NodeID) (_ map[storj.NodeID]*SelectedNode, err error) {
func (service *Service) CachedGetOnlineNodesForGet(ctx context.Context, nodeIDs []storj.NodeID) (_ map[storj.NodeID]*uploadselection.SelectedNode, err error) {
defer mon.Task()(&ctx)(&err)
return service.DownloadSelectionCache.GetNodes(ctx, nodeIDs)
}
@ -453,7 +433,7 @@ func (service *Service) GetNodesOutOfPlacement(ctx context.Context, nodeIDs []st
}
// FindStorageNodesForGracefulExit searches the overlay network for nodes that meet the provided requirements for graceful-exit requests.
func (service *Service) FindStorageNodesForGracefulExit(ctx context.Context, req FindStorageNodesRequest) (_ []*SelectedNode, err error) {
func (service *Service) FindStorageNodesForGracefulExit(ctx context.Context, req FindStorageNodesRequest) (_ []*uploadselection.SelectedNode, err error) {
defer mon.Task()(&ctx)(&err)
return service.UploadSelectionCache.GetNodes(ctx, req)
}
@ -462,7 +442,7 @@ func (service *Service) FindStorageNodesForGracefulExit(ctx context.Context, req
//
// When enabled it uses the cache to select nodes.
// When the node selection from the cache fails, it falls back to the old implementation.
func (service *Service) FindStorageNodesForUpload(ctx context.Context, req FindStorageNodesRequest) (_ []*SelectedNode, err error) {
func (service *Service) FindStorageNodesForUpload(ctx context.Context, req FindStorageNodesRequest) (_ []*uploadselection.SelectedNode, err error) {
defer mon.Task()(&ctx)(&err)
if service.config.Node.AsOfSystemTime.Enabled && service.config.Node.AsOfSystemTime.DefaultInterval < 0 {
req.AsOfSystemInterval = service.config.Node.AsOfSystemTime.DefaultInterval
@ -498,7 +478,7 @@ func (service *Service) FindStorageNodesForUpload(ctx context.Context, req FindS
// FindStorageNodesWithPreferences searches the overlay network for nodes that meet the provided criteria.
//
// This does not use a cache.
func (service *Service) FindStorageNodesWithPreferences(ctx context.Context, req FindStorageNodesRequest, preferences *NodeSelectionConfig) (nodes []*SelectedNode, err error) {
func (service *Service) FindStorageNodesWithPreferences(ctx context.Context, req FindStorageNodesRequest, preferences *NodeSelectionConfig) (nodes []*uploadselection.SelectedNode, err error) {
defer mon.Task()(&ctx)(&err)
// TODO: add sanity limits to requested node count
// TODO: add sanity limits to excluded nodes
@ -584,7 +564,7 @@ func (service *Service) KnownReliableInExcludedCountries(ctx context.Context, no
}
// KnownReliable filters a set of nodes to reliable (online and qualified) nodes.
func (service *Service) KnownReliable(ctx context.Context, nodeIDs storj.NodeIDList) (onlineNodes []SelectedNode, offlineNodes []SelectedNode, err error) {
func (service *Service) KnownReliable(ctx context.Context, nodeIDs storj.NodeIDList) (onlineNodes []uploadselection.SelectedNode, offlineNodes []uploadselection.SelectedNode, err error) {
defer mon.Task()(&ctx)(&err)
// TODO add as of system time
@ -840,7 +820,7 @@ func (service *Service) DisqualifyNode(ctx context.Context, nodeID storj.NodeID,
}
// SelectAllStorageNodesDownload returns a nodes that are ready for downloading.
func (service *Service) SelectAllStorageNodesDownload(ctx context.Context, onlineWindow time.Duration, asOf AsOfSystemTimeConfig) (_ []*SelectedNode, err error) {
func (service *Service) SelectAllStorageNodesDownload(ctx context.Context, onlineWindow time.Duration, asOf AsOfSystemTimeConfig) (_ []*uploadselection.SelectedNode, err error) {
defer mon.Task()(&ctx)(&err)
return service.db.SelectAllStorageNodesDownload(ctx, onlineWindow, asOf)
}

7
satellite/overlay/service_test.go
View File

@ -24,6 +24,7 @@ import (
"storj.io/storj/private/testplanet"
"storj.io/storj/satellite"
"storj.io/storj/satellite/nodeevents"
"storj.io/storj/satellite/nodeselection/uploadselection"
"storj.io/storj/satellite/overlay"
"storj.io/storj/satellite/reputation"
"storj.io/storj/satellite/satellitedb/satellitedbtest"
@ -205,7 +206,7 @@ func TestRandomizedSelection(t *testing.T) {
// select numNodesToSelect nodes selectIterations times
for i := 0; i < selectIterations; i++ {
var nodes []*overlay.SelectedNode
var nodes []*uploadselection.SelectedNode
var err error
if i%2 == 0 {
@ -326,7 +327,7 @@ func TestRandomizedSelectionCache(t *testing.T) {
// select numNodesToSelect nodes selectIterations times
for i := 0; i < selectIterations; i++ {
var nodes []*overlay.SelectedNode
var nodes []*uploadselection.SelectedNode
var err error
req := overlay.FindStorageNodesRequest{
RequestedCount: numNodesToSelect,
@ -670,7 +671,7 @@ func TestSuspendedSelection(t *testing.T) {
}
}
var nodes []*overlay.SelectedNode
var nodes []*uploadselection.SelectedNode
var err error
numNodesToSelect := 10

5
satellite/overlay/statdb_test.go
View File

@ -18,6 +18,7 @@ import (
"storj.io/common/storj/location"
"storj.io/common/testcontext"
"storj.io/storj/satellite"
"storj.io/storj/satellite/nodeselection/uploadselection"
"storj.io/storj/satellite/overlay"
"storj.io/storj/satellite/satellitedb/satellitedbtest"
)
@ -106,8 +107,8 @@ func testDatabase(ctx context.Context, t *testing.T, cache overlay.DB) {
ExcludedCountries: []string{"FR", "BE"},
}
contains := func(nodeID storj.NodeID) func(node overlay.SelectedNode) bool {
return func(node overlay.SelectedNode) bool {
contains := func(nodeID storj.NodeID) func(node uploadselection.SelectedNode) bool {
return func(node uploadselection.SelectedNode) bool {
return node.ID == nodeID
}
}

37
satellite/overlay/uploadselection.go
View File

@ -9,7 +9,6 @@ import (
"go.uber.org/zap"
"storj.io/common/pb"
"storj.io/common/sync2"
"storj.io/storj/satellite/nodeselection/uploadselection"
)
@ -19,7 +18,7 @@ import (
// architecture: Database
type UploadSelectionDB interface {
// SelectAllStorageNodesUpload returns all nodes that qualify to store data, organized as reputable nodes and new nodes
SelectAllStorageNodesUpload(ctx context.Context, selectionCfg NodeSelectionConfig) (reputable, new []*SelectedNode, err error)
SelectAllStorageNodesUpload(ctx context.Context, selectionCfg NodeSelectionConfig) (reputable, new []*uploadselection.SelectedNode, err error)
}
// UploadSelectionCacheConfig is a configuration for upload selection cache.
@ -73,7 +72,7 @@ func (cache *UploadSelectionCache) read(ctx context.Context) (_ *uploadselection
return nil, Error.Wrap(err)
}
state := uploadselection.NewState(convSelectedNodesToNodes(reputableNodes), convSelectedNodesToNodes(newNodes))
state := uploadselection.NewState(reputableNodes, newNodes)
mon.IntVal("refresh_cache_size_reputable").Observe(int64(len(reputableNodes)))
mon.IntVal("refresh_cache_size_new").Observe(int64(len(newNodes)))
@ -84,7 +83,7 @@ func (cache *UploadSelectionCache) read(ctx context.Context) (_ *uploadselection
// GetNodes selects nodes from the cache that will be used to upload a file.
// Every node selected will be from a distinct network.
// If the cache hasn't been refreshed recently it will do so first.
func (cache *UploadSelectionCache) GetNodes(ctx context.Context, req FindStorageNodesRequest) (_ []*SelectedNode, err error) {
func (cache *UploadSelectionCache) GetNodes(ctx context.Context, req FindStorageNodesRequest) (_ []*uploadselection.SelectedNode, err error) {
defer mon.Task()(&ctx)(&err)
state, err := cache.cache.Get(ctx, time.Now())
@ -103,7 +102,7 @@ func (cache *UploadSelectionCache) GetNodes(ctx context.Context, req FindStorage
err = ErrNotEnoughNodes.Wrap(err)
}
return convNodesToSelectedNodes(selected), err
return selected, err
}
// Size returns how many reputable nodes and new nodes are in the cache.
@ -115,31 +114,3 @@ func (cache *UploadSelectionCache) Size(ctx context.Context) (reputableNodeCount
stats := state.Stats()
return stats.Reputable, stats.New, nil
}
func convNodesToSelectedNodes(nodes []*uploadselection.Node) (xs []*SelectedNode) {
for _, n := range nodes {
xs = append(xs, &SelectedNode{
ID: n.ID,
Address: pb.NodeFromNodeURL(n.NodeURL).Address,
LastNet: n.LastNet,
LastIPPort: n.LastIPPort,
CountryCode: n.CountryCode,
})
}
return xs
}
func convSelectedNodesToNodes(nodes []*SelectedNode) (xs []*uploadselection.Node) {
for _, n := range nodes {
xs = append(xs, &uploadselection.Node{
NodeURL: (&pb.Node{
Id: n.ID,
Address: n.Address,
}).NodeURL(),
LastNet: n.LastNet,
LastIPPort: n.LastIPPort,
CountryCode: n.CountryCode,
})
}
return xs
}

21
satellite/overlay/uploadselection_test.go
View File

@ -22,6 +22,7 @@ import (
"storj.io/common/testrand"
"storj.io/storj/private/testplanet"
"storj.io/storj/satellite"
"storj.io/storj/satellite/nodeselection/uploadselection"
"storj.io/storj/satellite/overlay"
"storj.io/storj/satellite/satellitedb/satellitedbtest"
)
@ -126,21 +127,21 @@ func addNodesToNodesTable(ctx context.Context, t *testing.T, db overlay.DB, coun
type mockdb struct {
mu sync.Mutex
callCount int
reputable []*overlay.SelectedNode
new []*overlay.SelectedNode
reputable []*uploadselection.SelectedNode
new []*uploadselection.SelectedNode
}
func (m *mockdb) SelectAllStorageNodesUpload(ctx context.Context, selectionCfg overlay.NodeSelectionConfig) (reputable, new []*overlay.SelectedNode, err error) {
func (m *mockdb) SelectAllStorageNodesUpload(ctx context.Context, selectionCfg overlay.NodeSelectionConfig) (reputable, new []*uploadselection.SelectedNode, err error) {
m.mu.Lock()
defer m.mu.Unlock()
sync2.Sleep(ctx, 500*time.Millisecond)
m.callCount++
reputable = make([]*overlay.SelectedNode, len(m.reputable))
reputable = make([]*uploadselection.SelectedNode, len(m.reputable))
for i, n := range m.reputable {
reputable[i] = n.Clone()
}
new = make([]*overlay.SelectedNode, len(m.new))
new = make([]*uploadselection.SelectedNode, len(m.new))
for i, n := range m.new {
new[i] = n.Clone()
}
@ -275,13 +276,13 @@ func TestGetNodesConcurrent(t *testing.T) {
ctx := testcontext.New(t)
defer ctx.Cleanup()
reputableNodes := []*overlay.SelectedNode{{
reputableNodes := []*uploadselection.SelectedNode{{
ID: storj.NodeID{1},
Address: &pb.NodeAddress{Address: "127.0.0.9"},
LastNet: "127.0.0",
LastIPPort: "127.0.0.9:8000",
}}
newNodes := []*overlay.SelectedNode{{
newNodes := []*uploadselection.SelectedNode{{
ID: storj.NodeID{1},
Address: &pb.NodeAddress{Address: "127.0.0.10"},
LastNet: "127.0.0",
@ -380,7 +381,7 @@ func TestGetNodesDistinct(t *testing.T) {
ctx := testcontext.New(t)
defer ctx.Cleanup()
reputableNodes := []*overlay.SelectedNode{{
reputableNodes := []*uploadselection.SelectedNode{{
ID: testrand.NodeID(),
Address: &pb.NodeAddress{Address: "127.0.0.9"},
LastNet: "127.0.0",
@ -402,7 +403,7 @@ func TestGetNodesDistinct(t *testing.T) {
LastIPPort: "127.0.2.7:8000",
}}
newNodes := []*overlay.SelectedNode{{
newNodes := []*uploadselection.SelectedNode{{
ID: testrand.NodeID(),
Address: &pb.NodeAddress{Address: "127.0.0.10"},
LastNet: "127.0.0",
@ -460,7 +461,7 @@ func TestGetNodesDistinct(t *testing.T) {
{ // test that distinctIP=true allows selecting 6 nodes
// emulate DistinctIP=false behavior by filling in LastNets with unique addresses
for _, nodeList := range [][]*overlay.SelectedNode{reputableNodes, newNodes} {
for _, nodeList := range [][]*uploadselection.SelectedNode{reputableNodes, newNodes} {
for i := range nodeList {
nodeList[i].LastNet = nodeList[i].LastIPPort
}

5
satellite/repair/repairer/segments.go
View File

@ -21,6 +21,7 @@ import (
"storj.io/common/sync2"
"storj.io/storj/satellite/audit"
"storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/nodeselection/uploadselection"
"storj.io/storj/satellite/orders"
"storj.io/storj/satellite/overlay"
"storj.io/storj/satellite/repair"
@ -681,7 +682,7 @@ func (repairer *SegmentRepairer) classifySegmentPieces(ctx context.Context, segm
reliablePieces := metabase.Pieces{}
collectLastNets := func(reliable []overlay.SelectedNode) {
collectLastNets := func(reliable []uploadselection.SelectedNode) {
for _, node := range reliable {
pieceNum := nodeIDPieceMap[node.ID]
reliablePieces = append(reliablePieces, metabase.Piece{
@ -704,7 +705,7 @@ func (repairer *SegmentRepairer) classifySegmentPieces(ctx context.Context, segm
if repairer.doPlacementCheck && segment.Placement != storj.EveryCountry {
result.OutOfPlacementPiecesSet = map[uint16]bool{}
checkPlacement := func(reliable []overlay.SelectedNode) {
checkPlacement := func(reliable []uploadselection.SelectedNode) {
for _, node := range reliable {
if segment.Placement.AllowedCountry(node.CountryCode) {
continue

7
satellite/satellitedb/nodeselection.go
View File

@ -16,10 +16,11 @@ import (
"storj.io/common/storj"
"storj.io/private/dbutil/pgutil"
"storj.io/private/version"
"storj.io/storj/satellite/nodeselection/uploadselection"
"storj.io/storj/satellite/overlay"
)
func (cache *overlaycache) SelectStorageNodes(ctx context.Context, totalNeededNodes, newNodeCount int, criteria *overlay.NodeCriteria) (nodes []*overlay.SelectedNode, err error) {
func (cache *overlaycache) SelectStorageNodes(ctx context.Context, totalNeededNodes, newNodeCount int, criteria *overlay.NodeCriteria) (nodes []*uploadselection.SelectedNode, err error) {
defer mon.Task()(&ctx)(&err)
if totalNeededNodes == 0 {
return nil, nil
@ -86,7 +87,7 @@ func (cache *overlaycache) SelectStorageNodes(ctx context.Context, totalNeededNo
return nodes, nil
}
func (cache *overlaycache) selectStorageNodesOnce(ctx context.Context, reputableNodeCount, newNodeCount int, criteria *overlay.NodeCriteria, excludedIDs []storj.NodeID, excludedNetworks []string) (reputableNodes, newNodes []*overlay.SelectedNode, err error) {
func (cache *overlaycache) selectStorageNodesOnce(ctx context.Context, reputableNodeCount, newNodeCount int, criteria *overlay.NodeCriteria, excludedIDs []storj.NodeID, excludedNetworks []string) (reputableNodes, newNodes []*uploadselection.SelectedNode, err error) {
defer mon.Task()(&ctx)(&err)
newNodesCondition, err := nodeSelectionCondition(ctx, criteria, excludedIDs, excludedNetworks, true)
@ -127,7 +128,7 @@ func (cache *overlaycache) selectStorageNodesOnce(ctx context.Context, reputable
defer func() { err = errs.Combine(err, rows.Close()) }()
for rows.Next() {
var node overlay.SelectedNode
var node uploadselection.SelectedNode
node.Address = &pb.NodeAddress{}
var lastIPPort sql.NullString
var isNew bool

29
satellite/satellitedb/overlaycache.go
View File

@ -23,6 +23,7 @@ import (
"storj.io/private/dbutil/pgutil"
"storj.io/private/tagsql"
"storj.io/private/version"
"storj.io/storj/satellite/nodeselection/uploadselection"
"storj.io/storj/satellite/overlay"
"storj.io/storj/satellite/satellitedb/dbx"
)
@ -38,7 +39,7 @@ type overlaycache struct {
}
// SelectAllStorageNodesUpload returns all nodes that qualify to store data, organized as reputable nodes and new nodes.
func (cache *overlaycache) SelectAllStorageNodesUpload(ctx context.Context, selectionCfg overlay.NodeSelectionConfig) (reputable, new []*overlay.SelectedNode, err error) {
func (cache *overlaycache) SelectAllStorageNodesUpload(ctx context.Context, selectionCfg overlay.NodeSelectionConfig) (reputable, new []*uploadselection.SelectedNode, err error) {
for {
reputable, new, err = cache.selectAllStorageNodesUpload(ctx, selectionCfg)
if err != nil {
@ -53,7 +54,7 @@ func (cache *overlaycache) SelectAllStorageNodesUpload(ctx context.Context, sele
return reputable, new, err
}
func (cache *overlaycache) selectAllStorageNodesUpload(ctx context.Context, selectionCfg overlay.NodeSelectionConfig) (reputable, new []*overlay.SelectedNode, err error) {
func (cache *overlaycache) selectAllStorageNodesUpload(ctx context.Context, selectionCfg overlay.NodeSelectionConfig) (reputable, new []*uploadselection.SelectedNode, err error) {
defer mon.Task()(&ctx)(&err)
query := `
@ -94,10 +95,10 @@ func (cache *overlaycache) selectAllStorageNodesUpload(ctx context.Context, sele
}
defer func() { err = errs.Combine(err, rows.Close()) }()
var reputableNodes []*overlay.SelectedNode
var newNodes []*overlay.SelectedNode
var reputableNodes []*uploadselection.SelectedNode
var newNodes []*uploadselection.SelectedNode
for rows.Next() {
var node overlay.SelectedNode
var node uploadselection.SelectedNode
node.Address = &pb.NodeAddress{}
var lastIPPort sql.NullString
var vettedAt *time.Time
@ -123,7 +124,7 @@ func (cache *overlaycache) selectAllStorageNodesUpload(ctx context.Context, sele
}
// SelectAllStorageNodesDownload returns all nodes that qualify to store data, organized as reputable nodes and new nodes.
func (cache *overlaycache) SelectAllStorageNodesDownload(ctx context.Context, onlineWindow time.Duration, asOf overlay.AsOfSystemTimeConfig) (nodes []*overlay.SelectedNode, err error) {
func (cache *overlaycache) SelectAllStorageNodesDownload(ctx context.Context, onlineWindow time.Duration, asOf overlay.AsOfSystemTimeConfig) (nodes []*uploadselection.SelectedNode, err error) {
for {
nodes, err = cache.selectAllStorageNodesDownload(ctx, onlineWindow, asOf)
if err != nil {
@ -138,7 +139,7 @@ func (cache *overlaycache) SelectAllStorageNodesDownload(ctx context.Context, on
return nodes, err
}
func (cache *overlaycache) selectAllStorageNodesDownload(ctx context.Context, onlineWindow time.Duration, asOfConfig overlay.AsOfSystemTimeConfig) (_ []*overlay.SelectedNode, err error) {
func (cache *overlaycache) selectAllStorageNodesDownload(ctx context.Context, onlineWindow time.Duration, asOfConfig overlay.AsOfSystemTimeConfig) (_ []*uploadselection.SelectedNode, err error) {
defer mon.Task()(&ctx)(&err)
query := `
@ -160,9 +161,9 @@ func (cache *overlaycache) selectAllStorageNodesDownload(ctx context.Context, on
}
defer func() { err = errs.Combine(err, rows.Close()) }()
var nodes []*overlay.SelectedNode
var nodes []*uploadselection.SelectedNode
for rows.Next() {
var node overlay.SelectedNode
var node uploadselection.SelectedNode
node.Address = &pb.NodeAddress{}
var lastIPPort sql.NullString
var noise noiseScanner
@ -448,7 +449,7 @@ func (cache *overlaycache) knownReliableInExcludedCountries(ctx context.Context,
}
// KnownReliable filters a set of nodes to reliable nodes. List is split into online and offline nodes.
func (cache *overlaycache) KnownReliable(ctx context.Context, nodeIDs storj.NodeIDList, onlineWindow, asOfSystemInterval time.Duration) (online []overlay.SelectedNode, offline []overlay.SelectedNode, err error) {
func (cache *overlaycache) KnownReliable(ctx context.Context, nodeIDs storj.NodeIDList, onlineWindow, asOfSystemInterval time.Duration) (online []uploadselection.SelectedNode, offline []uploadselection.SelectedNode, err error) {
for {
online, offline, err = cache.knownReliable(ctx, nodeIDs, onlineWindow, asOfSystemInterval)
if err != nil {
@ -463,7 +464,7 @@ func (cache *overlaycache) KnownReliable(ctx context.Context, nodeIDs storj.Node
return online, offline, err
}
func (cache *overlaycache) knownReliable(ctx context.Context, nodeIDs storj.NodeIDList, onlineWindow, asOfSystemInterval time.Duration) (online []overlay.SelectedNode, offline []overlay.SelectedNode, err error) {
func (cache *overlaycache) knownReliable(ctx context.Context, nodeIDs storj.NodeIDList, onlineWindow, asOfSystemInterval time.Duration) (online []uploadselection.SelectedNode, offline []uploadselection.SelectedNode, err error) {
defer mon.Task()(&ctx)(&err)
if len(nodeIDs) == 0 {
@ -483,7 +484,7 @@ func (cache *overlaycache) knownReliable(ctx context.Context, nodeIDs storj.Node
))(func(rows tagsql.Rows) error {
for rows.Next() {
var onlineNode bool
var node overlay.SelectedNode
var node uploadselection.SelectedNode
node.Address = &pb.NodeAddress{}
var lastIPPort sql.NullString
err = rows.Scan(&node.ID, &node.Address.Address, &node.LastNet, &lastIPPort, &node.CountryCode, &onlineNode)
@ -1472,7 +1473,7 @@ func (cache *overlaycache) TestNodeCountryCode(ctx context.Context, nodeID storj
}
// IterateAllContactedNodes will call cb on all known nodes (used in restore trash contexts).
func (cache *overlaycache) IterateAllContactedNodes(ctx context.Context, cb func(context.Context, *overlay.SelectedNode) error) (err error) {
func (cache *overlaycache) IterateAllContactedNodes(ctx context.Context, cb func(context.Context, *uploadselection.SelectedNode) error) (err error) {
defer mon.Task()(&ctx)(&err)
var rows tagsql.Rows
@ -1488,7 +1489,7 @@ func (cache *overlaycache) IterateAllContactedNodes(ctx context.Context, cb func
defer func() { err = errs.Combine(err, rows.Close()) }()
for rows.Next() {
var node overlay.SelectedNode
var node uploadselection.SelectedNode
node.Address = &pb.NodeAddress{}
var lastIPPort sql.NullString

21
satellite/satellitedb/overlaycache_test.go
View File

@ -21,6 +21,7 @@ import (
"storj.io/private/version"
"storj.io/storj/private/teststorj"
"storj.io/storj/satellite"
"storj.io/storj/satellite/nodeselection/uploadselection"
"storj.io/storj/satellite/overlay"
"storj.io/storj/satellite/satellitedb/satellitedbtest"
)
@ -386,8 +387,8 @@ func TestOverlayCache_SelectAllStorageNodesDownloadUpload(t *testing.T) {
require.NoError(t, err)
}
checkNodes := func(selectedNodes []*overlay.SelectedNode) {
selectedNodesMap := map[storj.NodeID]*overlay.SelectedNode{}
checkNodes := func(selectedNodes []*uploadselection.SelectedNode) {
selectedNodesMap := map[storj.NodeID]*uploadselection.SelectedNode{}
for _, node := range selectedNodes {
selectedNodesMap[node.ID] = node
}
@ -423,7 +424,7 @@ func TestOverlayCache_KnownReliable(t *testing.T) {
satellitedbtest.Run(t, func(ctx *testcontext.Context, t *testing.T, db satellite.DB) {
cache := db.OverlayCache()
allNodes := []overlay.SelectedNode{
allNodes := []uploadselection.SelectedNode{
addNode(ctx, t, cache, "online", "127.0.0.1", true, false, false, false, false),
addNode(ctx, t, cache, "offline", "127.0.0.2", false, false, false, false, false),
addNode(ctx, t, cache, "disqalified", "127.0.0.3", false, true, false, false, false),
@ -432,7 +433,7 @@ func TestOverlayCache_KnownReliable(t *testing.T) {
addNode(ctx, t, cache, "exited", "127.0.0.6", false, false, false, false, true),
}
ids := func(nodes ...overlay.SelectedNode) storj.NodeIDList {
ids := func(nodes ...uploadselection.SelectedNode) storj.NodeIDList {
nodeIds := storj.NodeIDList{}
for _, node := range nodes {
nodeIds = append(nodeIds, node.ID)
@ -440,14 +441,14 @@ func TestOverlayCache_KnownReliable(t *testing.T) {
return nodeIds
}
nodes := func(nodes ...overlay.SelectedNode) []overlay.SelectedNode {
return append([]overlay.SelectedNode{}, nodes...)
nodes := func(nodes ...uploadselection.SelectedNode) []uploadselection.SelectedNode {
return append([]uploadselection.SelectedNode{}, nodes...)
}
type testCase struct {
IDs storj.NodeIDList
Online []overlay.SelectedNode
Offline []overlay.SelectedNode
Online []uploadselection.SelectedNode
Offline []uploadselection.SelectedNode
}
shuffledNodeIDs := ids(allNodes...)
@ -500,8 +501,8 @@ func TestOverlayCache_KnownReliable(t *testing.T) {
})
}
func addNode(ctx context.Context, t *testing.T, cache overlay.DB, address, lastIPPort string, online, disqalified, auditSuspended, offlineSuspended, exited bool) overlay.SelectedNode {
selectedNode := overlay.SelectedNode{
func addNode(ctx context.Context, t *testing.T, cache overlay.DB, address, lastIPPort string, online, disqalified, auditSuspended, offlineSuspended, exited bool) uploadselection.SelectedNode {
selectedNode := uploadselection.SelectedNode{
ID: testrand.NodeID(),
Address: &pb.NodeAddress{Address: address},
LastNet: lastIPPort,

6
storagenode/orders/service_test.go
View File

@ -16,7 +16,7 @@ import (
"storj.io/common/testrand"
"storj.io/storj/private/testplanet"
"storj.io/storj/satellite/metabase"
"storj.io/storj/satellite/overlay"
"storj.io/storj/satellite/nodeselection/uploadselection"
"storj.io/storj/storagenode"
"storj.io/storj/storagenode/orders"
"storj.io/storj/storagenode/orders/ordersfile"
@ -41,7 +41,7 @@ func TestOrderDBSettle(t *testing.T) {
_, orderLimits, piecePrivateKey, err := satellite.Orders.Service.CreatePutOrderLimits(
ctx,
metabase.BucketLocation{ProjectID: planet.Uplinks[0].Projects[0].ID, BucketName: bucketname},
[]*overlay.SelectedNode{
[]*uploadselection.SelectedNode{
{ID: node.ID(), LastIPPort: "fake", Address: new(pb.NodeAddress)},
},
time.Now().Add(2*time.Hour),
@ -147,7 +147,7 @@ func TestOrderFileStoreAndDBSettle(t *testing.T) {
_, orderLimits, piecePrivateKey, err := satellite.Orders.Service.CreatePutOrderLimits(
ctx,
metabase.BucketLocation{ProjectID: uplinkPeer.Projects[0].ID, BucketName: bucketname},
[]*overlay.SelectedNode{
[]*uploadselection.SelectedNode{
{ID: node.ID(), LastIPPort: "fake", Address: new(pb.NodeAddress)},
},
time.Now().Add(2*time.Hour),