storj/storagenode/storagenodedb/storagenodedbtest/run_test.go
Jennifer Li Johnson 724bb44723
Remove Kademlia dependencies from Satellite and Storagenode (#2966)
What:

cmd/inspector/main.go: removes kad commands
internal/testplanet/planet.go: Waits for contact chore to finish
satellite/contact/nodesservice.go: creates an empty nodes service implementation
satellite/contact/service.go: implements Local and FetchInfo methods & adds external address config value
satellite/discovery/service.go: replaces kad.FetchInfo with contact.FetchInfo in Refresh() & removes Discover()
satellite/peer.go: sets up contact service and endpoints
storagenode/console/service.go: replaces nodeID with contact.Local()
storagenode/contact/chore.go: replaces routing table with contact service
storagenode/contact/nodesservice.go: creates empty implementation for ping and request info nodes service & implements RequestInfo method
storagenode/contact/service.go: creates a service to return the local node and update its own capacity
storagenode/monitor/monitor.go: uses contact service in place of routing table
storagenode/operator.go: moves operatorconfig from kad into its own setup
storagenode/peer.go: sets up contact service, chore, pingstats and endpoints
satellite/overlay/config.go: changes NodeSelectionConfig.OnlineWindow default to 4hr to allow for accurate repair selection
Removes kademlia setups in:

cmd/storagenode/main.go
cmd/storj-sim/network.go
internal/testplane/planet.go
internal/testplanet/satellite.go
internal/testplanet/storagenode.go
satellite/peer.go
scripts/test-sim-backwards.sh
scripts/testdata/satellite-config.yaml.lock
storagenode/inspector/inspector.go
storagenode/peer.go
storagenode/storagenodedb/database.go
Why: Replacing Kademlia

Please describe the tests:
• internal/testplanet/planet_test.go:

TestBasic: assert that the storagenode can check in with the satellite without any errors
TestContact: test that all nodes get inserted into both satellites' overlay cache during testplanet setup
• satellite/contact/contact_test.go:

TestFetchInfo: Tests that the FetchInfo method returns the correct info
• storagenode/contact/contact_test.go:

TestNodeInfoUpdated: tests that the contact chore updates the node information
TestRequestInfoEndpoint: tests that the Request info endpoint returns the correct info
Please describe the performance impact: Node discovery should be at least slightly more performant since each node connects directly to each satellite and no longer needs to wait for bootstrapping. It probably won't be faster in real time on start up since each node waits a random amount of time (less than 1 hr) to initialize its first connection (jitter).
2019-09-19 15:56:34 -04:00

176 lines
4.4 KiB
Go

// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.
package storagenodedbtest_test
import (
"path/filepath"
"runtime"
"sync"
"testing"
"time"
"github.com/skyrings/skyring-common/tools/uuid"
"github.com/stretchr/testify/require"
"go.uber.org/zap/zaptest"
"storj.io/storj/internal/testcontext"
"storj.io/storj/internal/testidentity"
"storj.io/storj/internal/testrand"
"storj.io/storj/pkg/pb"
"storj.io/storj/pkg/signing"
"storj.io/storj/pkg/storj"
"storj.io/storj/storagenode"
"storj.io/storj/storagenode/orders"
"storj.io/storj/storagenode/storagenodedb"
"storj.io/storj/storagenode/storagenodedb/storagenodedbtest"
)
func TestDatabase(t *testing.T) {
storagenodedbtest.Run(t, func(t *testing.T, db storagenode.DB) {
})
}
func TestFileConcurrency(t *testing.T) {
ctx := testcontext.New(t)
defer ctx.Cleanup()
log := zaptest.NewLogger(t)
db, err := storagenodedb.New(log, storagenodedb.Config{
Pieces: ctx.Dir("storage"),
Info2: ctx.Dir("storage") + "/info.db",
})
if err != nil {
t.Fatal(err)
}
defer ctx.Check(db.Close)
testConcurrency(t, ctx, db)
}
func TestInMemoryConcurrency(t *testing.T) {
ctx := testcontext.New(t)
defer ctx.Cleanup()
log := zaptest.NewLogger(t)
storageDir := ctx.Dir("storage")
cfg := storagenodedb.Config{
Pieces: storageDir,
Storage: storageDir,
Info: filepath.Join(storageDir, "piecestore.db"),
Info2: filepath.Join(storageDir, "info.db"),
}
db, err := storagenodedb.New(log, cfg)
if err != nil {
t.Fatal(err)
}
defer ctx.Check(db.Close)
testConcurrency(t, ctx, db)
}
func testConcurrency(t *testing.T, ctx *testcontext.Context, db *storagenodedb.DB) {
t.Run("Sqlite", func(t *testing.T) {
runtime.GOMAXPROCS(2)
err := db.CreateTables()
if err != nil {
t.Fatal(err)
}
ordersMap := make(map[string]orders.Info)
err = createOrders(t, ctx, ordersMap, 1000)
require.NoError(t, err)
err = insertOrders(t, ctx, db, ordersMap)
require.NoError(t, err)
err = verifyOrders(t, ctx, db, ordersMap)
require.NoError(t, err)
})
}
func insertOrders(t *testing.T, ctx *testcontext.Context, db *storagenodedb.DB, ordersMap map[string]orders.Info) (err error) {
var wg sync.WaitGroup
for _, order := range ordersMap {
wg.Add(1)
o := order
go insertOrder(t, ctx, db, &wg, &o)
}
wg.Wait()
return nil
}
func insertOrder(t *testing.T, ctx *testcontext.Context, db *storagenodedb.DB, wg *sync.WaitGroup, order *orders.Info) {
defer wg.Done()
err := db.Orders().Enqueue(ctx, order)
require.NoError(t, err)
}
func verifyOrders(t *testing.T, ctx *testcontext.Context, db *storagenodedb.DB, orders map[string]orders.Info) (err error) {
dbOrders, _ := db.Orders().ListUnsent(ctx, 10000)
found := 0
for _, order := range orders {
for _, dbOrder := range dbOrders {
if order.Order.SerialNumber == dbOrder.Order.SerialNumber {
//fmt.Printf("Found %v\n", order.Order.SerialNumber)
found++
}
}
}
require.Equal(t, len(orders), found, "Number found must equal the length of the test data")
return nil
}
func createOrders(t *testing.T, ctx *testcontext.Context, orders map[string]orders.Info, count int) (err error) {
for i := 0; i < count; i++ {
key, err := uuid.New()
if err != nil {
return err
}
order := createOrder(t, ctx)
orders[key.String()] = *order
}
return nil
}
func createOrder(t *testing.T, ctx *testcontext.Context) (info *orders.Info) {
storageNodeIdentity := testidentity.MustPregeneratedSignedIdentity(0, storj.LatestIDVersion())
satelliteIdentity := testidentity.MustPregeneratedSignedIdentity(1, storj.LatestIDVersion())
piecePublicKey, piecePrivateKey, err := storj.NewPieceKey()
require.NoError(t, err)
piece := testrand.PieceID()
serialNumber := testrand.SerialNumber()
expiration := time.Now()
limit, err := signing.SignOrderLimit(ctx, signing.SignerFromFullIdentity(satelliteIdentity), &pb.OrderLimit{
SerialNumber: serialNumber,
SatelliteId: satelliteIdentity.ID,
UplinkPublicKey: piecePublicKey,
StorageNodeId: storageNodeIdentity.ID,
PieceId: piece,
Limit: 100,
Action: pb.PieceAction_GET,
PieceExpiration: expiration,
OrderExpiration: expiration,
})
require.NoError(t, err)
order, err := signing.SignUplinkOrder(ctx, piecePrivateKey, &pb.Order{
SerialNumber: serialNumber,
Amount: 50,
})
require.NoError(t, err)
return &orders.Info{
Limit: limit,
Order: order,
}
}