aboutsummaryrefslogtreecommitdiffstats
path: root/swarm/network/stream/snapshot_retrieval_test.go
diff options
context:
space:
mode:
authorholisticode <holistic.computing@gmail.com>2018-07-31 04:55:25 +0800
committerBalint Gabor <balint.g@gmail.com>2018-07-31 04:55:25 +0800
commitd6efa691872efb723ea3177a92da9e9b31c34eba (patch)
tree9c7e85c9cab9a2cf1240db47a8de44162f69353e /swarm/network/stream/snapshot_retrieval_test.go
parent3ea8ac6a9ab9e56164707119e9142f06fae4c316 (diff)
downloaddexon-d6efa691872efb723ea3177a92da9e9b31c34eba.tar
dexon-d6efa691872efb723ea3177a92da9e9b31c34eba.tar.gz
dexon-d6efa691872efb723ea3177a92da9e9b31c34eba.tar.bz2
dexon-d6efa691872efb723ea3177a92da9e9b31c34eba.tar.lz
dexon-d6efa691872efb723ea3177a92da9e9b31c34eba.tar.xz
dexon-d6efa691872efb723ea3177a92da9e9b31c34eba.tar.zst
dexon-d6efa691872efb723ea3177a92da9e9b31c34eba.zip
Merge netsim mig to master (#17241)
* swarm: merged stream-tests migration to develop * swarm/network: expose simulation RandomUpNode to use in stream tests * swarm/network: wait for subs in PeerEvents and fix stream.runSyncTest * swarm: enforce waitkademlia for snapshot tests * swarm: fixed syncer tests and snapshot_sync_test * swarm: linting of simulation package * swarm: address review comments * swarm/network/stream: fix delivery_test bugs and refactor * swarm/network/stream: addressed PR comments @janos * swarm/network/stream: enforce waitKademlia, improve TestIntervals * swarm/network/stream: TestIntervals not waiting for chunk to be stored
Diffstat (limited to 'swarm/network/stream/snapshot_retrieval_test.go')
-rw-r--r--swarm/network/stream/snapshot_retrieval_test.go865
1 files changed, 230 insertions, 635 deletions
diff --git a/swarm/network/stream/snapshot_retrieval_test.go b/swarm/network/stream/snapshot_retrieval_test.go
index 9961a0bc7..4ff947b21 100644
--- a/swarm/network/stream/snapshot_retrieval_test.go
+++ b/swarm/network/stream/snapshot_retrieval_test.go
@@ -17,20 +17,19 @@ package stream
import (
"context"
- crand "crypto/rand"
"fmt"
- "math/rand"
- "strings"
+ "os"
"sync"
"testing"
"time"
- "github.com/ethereum/go-ethereum/common"
+ "github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p/discover"
- "github.com/ethereum/go-ethereum/p2p/simulations"
+ "github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/log"
"github.com/ethereum/go-ethereum/swarm/network"
- streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing"
+ "github.com/ethereum/go-ethereum/swarm/network/simulation"
+ "github.com/ethereum/go-ethereum/swarm/state"
"github.com/ethereum/go-ethereum/swarm/storage"
)
@@ -40,40 +39,6 @@ const (
maxFileSize = 40
)
-func initRetrievalTest() {
- //global func to get overlay address from discover ID
- toAddr = func(id discover.NodeID) *network.BzzAddr {
- addr := network.NewAddrFromNodeID(id)
- return addr
- }
- //global func to create local store
- createStoreFunc = createTestLocalStorageForId
- //local stores
- stores = make(map[discover.NodeID]storage.ChunkStore)
- //data directories for each node and store
- datadirs = make(map[discover.NodeID]string)
- //deliveries for each node
- deliveries = make(map[discover.NodeID]*Delivery)
- //global retrieve func
- getRetrieveFunc = func(id discover.NodeID) func(ctx context.Context, chunk *storage.Chunk) error {
- return func(ctx context.Context, chunk *storage.Chunk) error {
- skipCheck := true
- return deliveries[id].RequestFromPeers(ctx, chunk.Addr[:], skipCheck)
- }
- }
- //registries, map of discover.NodeID to its streamer
- registries = make(map[discover.NodeID]*TestRegistry)
- //not needed for this test but required from common_test for NewStreamService
- waitPeerErrC = make(chan error)
- //also not needed for this test but required for NewStreamService
- peerCount = func(id discover.NodeID) int {
- if ids[0] == id || ids[len(ids)-1] == id {
- return 1
- }
- return 2
- }
-}
-
//This test is a retrieval test for nodes.
//A configurable number of nodes can be
//provided to the test.
@@ -81,7 +46,10 @@ func initRetrievalTest() {
//Number of nodes can be provided via commandline too.
func TestFileRetrieval(t *testing.T) {
if *nodes != 0 {
- fileRetrievalTest(t, *nodes)
+ err := runFileRetrievalTest(*nodes)
+ if err != nil {
+ t.Fatal(err)
+ }
} else {
nodeCnt := []int{16}
//if the `longrunning` flag has been provided
@@ -90,7 +58,10 @@ func TestFileRetrieval(t *testing.T) {
nodeCnt = append(nodeCnt, 32, 64, 128)
}
for _, n := range nodeCnt {
- fileRetrievalTest(t, n)
+ err := runFileRetrievalTest(n)
+ if err != nil {
+ t.Fatal(err)
+ }
}
}
}
@@ -105,7 +76,10 @@ func TestRetrieval(t *testing.T) {
//if nodes/chunks have been provided via commandline,
//run the tests with these values
if *nodes != 0 && *chunks != 0 {
- retrievalTest(t, *chunks, *nodes)
+ err := runRetrievalTest(*chunks, *nodes)
+ if err != nil {
+ t.Fatal(err)
+ }
} else {
var nodeCnt []int
var chnkCnt []int
@@ -121,76 +95,17 @@ func TestRetrieval(t *testing.T) {
}
for _, n := range nodeCnt {
for _, c := range chnkCnt {
- retrievalTest(t, c, n)
+ err := runRetrievalTest(c, n)
+ if err != nil {
+ t.Fatal(err)
+ }
}
}
}
}
-//Every test runs 3 times, a live, a history, and a live AND history
-func fileRetrievalTest(t *testing.T, nodeCount int) {
- //test live and NO history
- log.Info("Testing live and no history", "nodeCount", nodeCount)
- live = true
- history = false
- err := runFileRetrievalTest(nodeCount)
- if err != nil {
- t.Fatal(err)
- }
- //test history only
- log.Info("Testing history only", "nodeCount", nodeCount)
- live = false
- history = true
- err = runFileRetrievalTest(nodeCount)
- if err != nil {
- t.Fatal(err)
- }
- //finally test live and history
- log.Info("Testing live and history", "nodeCount", nodeCount)
- live = true
- err = runFileRetrievalTest(nodeCount)
- if err != nil {
- t.Fatal(err)
- }
-}
-
-//Every test runs 3 times, a live, a history, and a live AND history
-func retrievalTest(t *testing.T, chunkCount int, nodeCount int) {
- //test live and NO history
- log.Info("Testing live and no history", "chunkCount", chunkCount, "nodeCount", nodeCount)
- live = true
- history = false
- err := runRetrievalTest(chunkCount, nodeCount)
- if err != nil {
- t.Fatal(err)
- }
- //test history only
- log.Info("Testing history only", "chunkCount", chunkCount, "nodeCount", nodeCount)
- live = false
- history = true
- err = runRetrievalTest(chunkCount, nodeCount)
- if err != nil {
- t.Fatal(err)
- }
- //finally test live and history
- log.Info("Testing live and history", "chunkCount", chunkCount, "nodeCount", nodeCount)
- live = true
- err = runRetrievalTest(chunkCount, nodeCount)
- if err != nil {
- t.Fatal(err)
- }
-}
-
/*
-The upload is done by dependency to the global
-`live` and `history` variables;
-
-If `live` is set, first stream subscriptions are established,
-then files are uploaded to nodes.
-
-If `history` is enabled, first upload files, then build up subscriptions.
-
The test loads a snapshot file to construct the swarm network,
assuming that the snapshot file identifies a healthy
kademlia network. Nevertheless a health check runs in the
@@ -199,261 +114,129 @@ simulation's `action` function.
The snapshot should have 'streamer' in its service list.
*/
func runFileRetrievalTest(nodeCount int) error {
- //for every run (live, history), int the variables
- initRetrievalTest()
- //the ids of the snapshot nodes, initiate only now as we need nodeCount
- ids = make([]discover.NodeID, nodeCount)
- //channel to check for disconnection errors
- disconnectC := make(chan error)
- //channel to close disconnection watcher routine
- quitC := make(chan struct{})
- //the test conf (using same as in `snapshot_sync_test`
- conf = &synctestConfig{}
- //map of overlay address to discover ID
- conf.addrToIdMap = make(map[string]discover.NodeID)
- //array where the generated chunk hashes will be stored
- conf.hashes = make([]storage.Address, 0)
- //load nodes from the snapshot file
- net, err := initNetWithSnapshot(nodeCount)
- if err != nil {
- return err
- }
- var rpcSubscriptionsWg sync.WaitGroup
- //do cleanup after test is terminated
- defer func() {
- //shutdown the snapshot network
- net.Shutdown()
- //after the test, clean up local stores initialized with createLocalStoreForId
- localStoreCleanup()
- //finally clear all data directories
- datadirsCleanup()
- }()
- //get the nodes of the network
- nodes := net.GetNodes()
- //iterate over all nodes...
- for c := 0; c < len(nodes); c++ {
- //create an array of discovery nodeIDS
- ids[c] = nodes[c].ID()
- a := network.ToOverlayAddr(ids[c].Bytes())
- //append it to the array of all overlay addresses
- conf.addrs = append(conf.addrs, a)
- conf.addrToIdMap[string(a)] = ids[c]
- }
+ sim := simulation.New(map[string]simulation.ServiceFunc{
+ "streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
- //needed for healthy call
- ppmap = network.NewPeerPotMap(testMinProxBinSize, conf.addrs)
-
- //an array for the random files
- var randomFiles []string
- //channel to signal when the upload has finished
- uploadFinished := make(chan struct{})
- //channel to trigger new node checks
- trigger := make(chan discover.NodeID)
- //simulation action
- action := func(ctx context.Context) error {
- //first run the health check on all nodes,
- //wait until nodes are all healthy
- ticker := time.NewTicker(200 * time.Millisecond)
- defer ticker.Stop()
- for range ticker.C {
- healthy := true
- for _, id := range ids {
- r := registries[id]
- //PeerPot for this node
- addr := common.Bytes2Hex(r.addr.OAddr)
- pp := ppmap[addr]
- //call Healthy RPC
- h := r.delivery.overlay.Healthy(pp)
- //print info
- log.Debug(r.delivery.overlay.String())
- log.Debug(fmt.Sprintf("IS HEALTHY: %t", h.GotNN && h.KnowNN && h.Full))
- if !h.GotNN || !h.Full {
- healthy = false
- break
- }
+ id := ctx.Config.ID
+ addr := network.NewAddrFromNodeID(id)
+ store, datadir, err := createTestLocalStorageForID(id, addr)
+ if err != nil {
+ return nil, nil, err
}
- if healthy {
- break
+ bucket.Store(bucketKeyStore, store)
+ cleanup = func() {
+ os.RemoveAll(datadir)
+ store.Close()
}
- }
+ localStore := store.(*storage.LocalStore)
+ db := storage.NewDBAPI(localStore)
+ kad := network.NewKademlia(addr.Over(), network.NewKadParams())
+ delivery := NewDelivery(kad, db)
- if history {
- log.Info("Uploading for history")
- //If testing only history, we upload the chunk(s) first
- conf.hashes, randomFiles, err = uploadFilesToNodes(nodes)
- if err != nil {
- return err
- }
- }
+ r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{
+ DoSync: true,
+ SyncUpdateDelay: 3 * time.Second,
+ })
- //variables needed to wait for all subscriptions established before uploading
- errc := make(chan error)
-
- //now setup and start event watching in order to know when we can upload
- ctx, watchCancel := context.WithTimeout(context.Background(), MaxTimeout*time.Second)
- defer watchCancel()
-
- log.Info("Setting up stream subscription")
- //We need two iterations, one to subscribe to the subscription events
- //(so we know when setup phase is finished), and one to
- //actually run the stream subscriptions. We can't do it in the same iteration,
- //because while the first nodes in the loop are setting up subscriptions,
- //the latter ones have not subscribed to listen to peer events yet,
- //and then we miss events.
-
- //first iteration: setup disconnection watcher and subscribe to peer events
- for j, id := range ids {
- log.Trace(fmt.Sprintf("Subscribe to subscription events: %d", j))
- client, err := net.GetNode(id).Client()
- if err != nil {
- return err
- }
- wsDoneC := watchSubscriptionEvents(ctx, id, client, errc, quitC)
- // doneC is nil, the error happened which is sent to errc channel, already
- if wsDoneC == nil {
- continue
- }
- rpcSubscriptionsWg.Add(1)
- go func() {
- <-wsDoneC
- rpcSubscriptionsWg.Done()
- }()
-
- //watch for peers disconnecting
- wdDoneC, err := streamTesting.WatchDisconnections(id, client, disconnectC, quitC)
- if err != nil {
- return err
- }
- rpcSubscriptionsWg.Add(1)
- go func() {
- <-wdDoneC
- rpcSubscriptionsWg.Done()
- }()
- }
+ fileStore := storage.NewFileStore(storage.NewNetStore(localStore, nil), storage.NewFileStoreParams())
+ bucket.Store(bucketKeyFileStore, fileStore)
- //second iteration: start syncing and setup stream subscriptions
- for j, id := range ids {
- log.Trace(fmt.Sprintf("Start syncing and stream subscriptions: %d", j))
- client, err := net.GetNode(id).Client()
- if err != nil {
- return err
- }
- //start syncing!
- var cnt int
- err = client.CallContext(ctx, &cnt, "stream_startSyncing")
- if err != nil {
- return err
- }
- //increment the number of subscriptions we need to wait for
- //by the count returned from startSyncing (SYNC subscriptions)
- subscriptionCount += cnt
- //now also add the number of RETRIEVAL_REQUEST subscriptions
- for snid := range registries[id].peers {
- subscriptionCount++
- err = client.CallContext(ctx, nil, "stream_subscribeStream", snid, NewStream(swarmChunkServerStreamName, "", false), nil, Top)
- if err != nil {
- return err
- }
- }
- }
+ return r, cleanup, nil
- //now wait until the number of expected subscriptions has been finished
- //`watchSubscriptionEvents` will write with a `nil` value to errc
- //every time a `SubscriptionMsg` has been received
- for err := range errc {
- if err != nil {
- return err
- }
- //`nil` received, decrement count
- subscriptionCount--
- //all subscriptions received
- if subscriptionCount == 0 {
- break
- }
- }
+ },
+ })
+ defer sim.Close()
- log.Info("Stream subscriptions successfully requested, action terminated")
+ log.Info("Initializing test config")
- if live {
- //upload generated files to nodes
- var hashes []storage.Address
- var rfiles []string
- hashes, rfiles, err = uploadFilesToNodes(nodes)
- if err != nil {
- return err
- }
- conf.hashes = append(conf.hashes, hashes...)
- randomFiles = append(randomFiles, rfiles...)
- //signal to the trigger loop that the upload has finished
- uploadFinished <- struct{}{}
- }
+ conf := &synctestConfig{}
+ //map of discover ID to indexes of chunks expected at that ID
+ conf.idToChunksMap = make(map[discover.NodeID][]int)
+ //map of overlay address to discover ID
+ conf.addrToIDMap = make(map[string]discover.NodeID)
+ //array where the generated chunk hashes will be stored
+ conf.hashes = make([]storage.Address, 0)
- return nil
+ err := sim.UploadSnapshot(fmt.Sprintf("testing/snapshot_%d.json", nodeCount))
+ if err != nil {
+ return err
}
- //check defines what will be checked during the test
- check := func(ctx context.Context, id discover.NodeID) (bool, error) {
-
- select {
- case <-ctx.Done():
- return false, ctx.Err()
- case e := <-disconnectC:
- log.Error(e.Error())
- return false, fmt.Errorf("Disconnect event detected, network unhealthy")
- default:
- }
- log.Trace(fmt.Sprintf("Checking node: %s", id))
- //if there are more than one chunk, test only succeeds if all expected chunks are found
- allSuccess := true
-
- //check on the node's FileStore (netstore)
- fileStore := registries[id].fileStore
- //check all chunks
- for i, hash := range conf.hashes {
- reader, _ := fileStore.Retrieve(context.TODO(), hash)
- //check that we can read the file size and that it corresponds to the generated file size
- if s, err := reader.Size(context.TODO(), nil); err != nil || s != int64(len(randomFiles[i])) {
- allSuccess = false
- log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id)
- } else {
- log.Debug(fmt.Sprintf("File with root hash %x successfully retrieved", hash))
- }
+ ctx, cancelSimRun := context.WithTimeout(context.Background(), 1*time.Minute)
+ defer cancelSimRun()
+
+ result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
+ nodeIDs := sim.UpNodeIDs()
+ for _, n := range nodeIDs {
+ //get the kademlia overlay address from this ID
+ a := network.ToOverlayAddr(n.Bytes())
+ //append it to the array of all overlay addresses
+ conf.addrs = append(conf.addrs, a)
+ //the proximity calculation is on overlay addr,
+ //the p2p/simulations check func triggers on discover.NodeID,
+ //so we need to know which overlay addr maps to which nodeID
+ conf.addrToIDMap[string(a)] = n
}
- return allSuccess, nil
- }
+ //an array for the random files
+ var randomFiles []string
+ //channel to signal when the upload has finished
+ //uploadFinished := make(chan struct{})
+ //channel to trigger new node checks
- //for each tick, run the checks on all nodes
- timingTicker := time.NewTicker(5 * time.Second)
- defer timingTicker.Stop()
- go func() {
- //for live upload, we should wait for uploads to have finished
- //before starting to trigger the checks, due to file size
- if live {
- <-uploadFinished
+ conf.hashes, randomFiles, err = uploadFilesToNodes(sim)
+ if err != nil {
+ return err
}
- for range timingTicker.C {
- for i := 0; i < len(ids); i++ {
- log.Trace(fmt.Sprintf("triggering step %d, id %s", i, ids[i]))
- trigger <- ids[i]
- }
+ if _, err := sim.WaitTillHealthy(ctx, 2); err != nil {
+ return err
}
- }()
-
- log.Info("Starting simulation run...")
-
- timeout := MaxTimeout * time.Second
- ctx, cancel := context.WithTimeout(context.Background(), timeout)
- defer cancel()
- //run the simulation
- result := simulations.NewSimulation(net).Run(ctx, &simulations.Step{
- Action: action,
- Trigger: trigger,
- Expect: &simulations.Expectation{
- Nodes: ids,
- Check: check,
- },
+ // File retrieval check is repeated until all uploaded files are retrieved from all nodes
+ // or until the timeout is reached.
+ allSuccess := false
+ for !allSuccess {
+ for _, id := range nodeIDs {
+ //for each expected chunk, check if it is in the local store
+ localChunks := conf.idToChunksMap[id]
+ localSuccess := true
+ for _, ch := range localChunks {
+ //get the real chunk by the index in the index array
+ chunk := conf.hashes[ch]
+ log.Trace(fmt.Sprintf("node has chunk: %s:", chunk))
+ //check if the expected chunk is indeed in the localstore
+ var err error
+ //check on the node's FileStore (netstore)
+ item, ok := sim.NodeItem(id, bucketKeyFileStore)
+ if !ok {
+ return fmt.Errorf("No registry")
+ }
+ fileStore := item.(*storage.FileStore)
+ //check all chunks
+ for i, hash := range conf.hashes {
+ reader, _ := fileStore.Retrieve(context.TODO(), hash)
+ //check that we can read the file size and that it corresponds to the generated file size
+ if s, err := reader.Size(ctx, nil); err != nil || s != int64(len(randomFiles[i])) {
+ allSuccess = false
+ log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id)
+ } else {
+ log.Debug(fmt.Sprintf("File with root hash %x successfully retrieved", hash))
+ }
+ }
+ if err != nil {
+ log.Warn(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id))
+ localSuccess = false
+ } else {
+ log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", chunk, id))
+ }
+ }
+ allSuccess = localSuccess
+ }
+ }
+ if !allSuccess {
+ return fmt.Errorf("Not all chunks succeeded!")
+ }
+ return nil
})
if result.Error != nil {
@@ -466,14 +249,6 @@ func runFileRetrievalTest(nodeCount int) error {
/*
The test generates the given number of chunks.
-The upload is done by dependency to the global
-`live` and `history` variables;
-
-If `live` is set, first stream subscriptions are established, then
-upload to a random node.
-
-If `history` is enabled, first upload then build up subscriptions.
-
The test loads a snapshot file to construct the swarm network,
assuming that the snapshot file identifies a healthy
kademlia network. Nevertheless a health check runs in the
@@ -482,259 +257,129 @@ simulation's `action` function.
The snapshot should have 'streamer' in its service list.
*/
func runRetrievalTest(chunkCount int, nodeCount int) error {
- //for every run (live, history), int the variables
- initRetrievalTest()
- //the ids of the snapshot nodes, initiate only now as we need nodeCount
- ids = make([]discover.NodeID, nodeCount)
- //channel to check for disconnection errors
- disconnectC := make(chan error)
- //channel to close disconnection watcher routine
- quitC := make(chan struct{})
- //the test conf (using same as in `snapshot_sync_test`
- conf = &synctestConfig{}
- //map of overlay address to discover ID
- conf.addrToIdMap = make(map[string]discover.NodeID)
- //array where the generated chunk hashes will be stored
- conf.hashes = make([]storage.Address, 0)
- //load nodes from the snapshot file
- net, err := initNetWithSnapshot(nodeCount)
- if err != nil {
- return err
- }
- var rpcSubscriptionsWg sync.WaitGroup
- //do cleanup after test is terminated
- defer func() {
- //shutdown the snapshot network
- net.Shutdown()
- //after the test, clean up local stores initialized with createLocalStoreForId
- localStoreCleanup()
- //finally clear all data directories
- datadirsCleanup()
- }()
- //get the nodes of the network
- nodes := net.GetNodes()
- //select one index at random...
- idx := rand.Intn(len(nodes))
- //...and get the the node at that index
- //this is the node selected for upload
- uploadNode := nodes[idx]
- //iterate over all nodes...
- for c := 0; c < len(nodes); c++ {
- //create an array of discovery nodeIDS
- ids[c] = nodes[c].ID()
- a := network.ToOverlayAddr(ids[c].Bytes())
- //append it to the array of all overlay addresses
- conf.addrs = append(conf.addrs, a)
- conf.addrToIdMap[string(a)] = ids[c]
- }
-
- //needed for healthy call
- ppmap = network.NewPeerPotMap(testMinProxBinSize, conf.addrs)
-
- trigger := make(chan discover.NodeID)
- //simulation action
- action := func(ctx context.Context) error {
- //first run the health check on all nodes,
- //wait until nodes are all healthy
- ticker := time.NewTicker(200 * time.Millisecond)
- defer ticker.Stop()
- for range ticker.C {
- healthy := true
- for _, id := range ids {
- r := registries[id]
- //PeerPot for this node
- addr := common.Bytes2Hex(network.ToOverlayAddr(id.Bytes()))
- pp := ppmap[addr]
- //call Healthy RPC
- h := r.delivery.overlay.Healthy(pp)
- //print info
- log.Debug(r.delivery.overlay.String())
- log.Debug(fmt.Sprintf("IS HEALTHY: %t", h.GotNN && h.KnowNN && h.Full))
- if !h.GotNN || !h.Full {
- healthy = false
- break
- }
- }
- if healthy {
- break
- }
- }
+ sim := simulation.New(map[string]simulation.ServiceFunc{
+ "streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
- if history {
- log.Info("Uploading for history")
- //If testing only history, we upload the chunk(s) first
- conf.hashes, err = uploadFileToSingleNodeStore(uploadNode.ID(), chunkCount)
+ id := ctx.Config.ID
+ addr := network.NewAddrFromNodeID(id)
+ store, datadir, err := createTestLocalStorageForID(id, addr)
if err != nil {
- return err
+ return nil, nil, err
}
- }
-
- //variables needed to wait for all subscriptions established before uploading
- errc := make(chan error)
-
- //now setup and start event watching in order to know when we can upload
- ctx, watchCancel := context.WithTimeout(context.Background(), MaxTimeout*time.Second)
- defer watchCancel()
-
- log.Info("Setting up stream subscription")
- //We need two iterations, one to subscribe to the subscription events
- //(so we know when setup phase is finished), and one to
- //actually run the stream subscriptions. We can't do it in the same iteration,
- //because while the first nodes in the loop are setting up subscriptions,
- //the latter ones have not subscribed to listen to peer events yet,
- //and then we miss events.
-
- //first iteration: setup disconnection watcher and subscribe to peer events
- for j, id := range ids {
- log.Trace(fmt.Sprintf("Subscribe to subscription events: %d", j))
- client, err := net.GetNode(id).Client()
- if err != nil {
- return err
+ bucket.Store(bucketKeyStore, store)
+ cleanup = func() {
+ os.RemoveAll(datadir)
+ store.Close()
}
+ localStore := store.(*storage.LocalStore)
+ db := storage.NewDBAPI(localStore)
+ kad := network.NewKademlia(addr.Over(), network.NewKadParams())
+ delivery := NewDelivery(kad, db)
- //check for `SubscribeMsg` events to know when setup phase is complete
- wsDoneC := watchSubscriptionEvents(ctx, id, client, errc, quitC)
- // doneC is nil, the error happened which is sent to errc channel, already
- if wsDoneC == nil {
- continue
- }
- rpcSubscriptionsWg.Add(1)
- go func() {
- <-wsDoneC
- rpcSubscriptionsWg.Done()
- }()
-
- //watch for peers disconnecting
- wdDoneC, err := streamTesting.WatchDisconnections(id, client, disconnectC, quitC)
- if err != nil {
- return err
- }
- rpcSubscriptionsWg.Add(1)
- go func() {
- <-wdDoneC
- rpcSubscriptionsWg.Done()
- }()
- }
+ r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{
+ DoSync: true,
+ SyncUpdateDelay: 0,
+ })
- //second iteration: start syncing and setup stream subscriptions
- for j, id := range ids {
- log.Trace(fmt.Sprintf("Start syncing and stream subscriptions: %d", j))
- client, err := net.GetNode(id).Client()
- if err != nil {
- return err
- }
- //start syncing!
- var cnt int
- err = client.CallContext(ctx, &cnt, "stream_startSyncing")
- if err != nil {
- return err
- }
- //increment the number of subscriptions we need to wait for
- //by the count returned from startSyncing (SYNC subscriptions)
- subscriptionCount += cnt
- //now also add the number of RETRIEVAL_REQUEST subscriptions
- for snid := range registries[id].peers {
- subscriptionCount++
- err = client.CallContext(ctx, nil, "stream_subscribeStream", snid, NewStream(swarmChunkServerStreamName, "", false), nil, Top)
- if err != nil {
- return err
- }
- }
- }
+ fileStore := storage.NewFileStore(storage.NewNetStore(localStore, nil), storage.NewFileStoreParams())
+ bucketKeyFileStore = simulation.BucketKey("filestore")
+ bucket.Store(bucketKeyFileStore, fileStore)
- //now wait until the number of expected subscriptions has been finished
- //`watchSubscriptionEvents` will write with a `nil` value to errc
- //every time a `SubscriptionMsg` has been received
- for err := range errc {
- if err != nil {
- return err
- }
- //`nil` received, decrement count
- subscriptionCount--
- //all subscriptions received
- if subscriptionCount == 0 {
- break
- }
- }
+ return r, cleanup, nil
- log.Info("Stream subscriptions successfully requested, action terminated")
+ },
+ })
+ defer sim.Close()
- if live {
- //now upload the chunks to the selected random single node
- chnks, err := uploadFileToSingleNodeStore(uploadNode.ID(), chunkCount)
- if err != nil {
- return err
- }
- conf.hashes = append(conf.hashes, chnks...)
- }
+ conf := &synctestConfig{}
+ //map of discover ID to indexes of chunks expected at that ID
+ conf.idToChunksMap = make(map[discover.NodeID][]int)
+ //map of overlay address to discover ID
+ conf.addrToIDMap = make(map[string]discover.NodeID)
+ //array where the generated chunk hashes will be stored
+ conf.hashes = make([]storage.Address, 0)
- return nil
+ err := sim.UploadSnapshot(fmt.Sprintf("testing/snapshot_%d.json", nodeCount))
+ if err != nil {
+ return err
}
- chunkSize := storage.DefaultChunkSize
-
- //check defines what will be checked during the test
- check := func(ctx context.Context, id discover.NodeID) (bool, error) {
-
- //don't check the uploader node
- if id == uploadNode.ID() {
- return true, nil
+ ctx := context.Background()
+ result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
+ nodeIDs := sim.UpNodeIDs()
+ for _, n := range nodeIDs {
+ //get the kademlia overlay address from this ID
+ a := network.ToOverlayAddr(n.Bytes())
+ //append it to the array of all overlay addresses
+ conf.addrs = append(conf.addrs, a)
+ //the proximity calculation is on overlay addr,
+ //the p2p/simulations check func triggers on discover.NodeID,
+ //so we need to know which overlay addr maps to which nodeID
+ conf.addrToIDMap[string(a)] = n
}
- select {
- case <-ctx.Done():
- return false, ctx.Err()
- case e := <-disconnectC:
- log.Error(e.Error())
- return false, fmt.Errorf("Disconnect event detected, network unhealthy")
- default:
+ //an array for the random files
+ var randomFiles []string
+ //this is the node selected for upload
+ node := sim.RandomUpNode()
+ item, ok := sim.NodeItem(node.ID, bucketKeyStore)
+ if !ok {
+ return fmt.Errorf("No localstore")
}
- log.Trace(fmt.Sprintf("Checking node: %s", id))
- //if there are more than one chunk, test only succeeds if all expected chunks are found
- allSuccess := true
-
- //check on the node's FileStore (netstore)
- fileStore := registries[id].fileStore
- //check all chunks
- for _, chnk := range conf.hashes {
- reader, _ := fileStore.Retrieve(context.TODO(), chnk)
- //assuming that reading the Size of the chunk is enough to know we found it
- if s, err := reader.Size(context.TODO(), nil); err != nil || s != chunkSize {
- allSuccess = false
- log.Warn("Retrieve error", "err", err, "chunk", chnk, "nodeId", id)
- } else {
- log.Debug(fmt.Sprintf("Chunk %x found", chnk))
- }
+ lstore := item.(*storage.LocalStore)
+ conf.hashes, err = uploadFileToSingleNodeStore(node.ID, chunkCount, lstore)
+ if err != nil {
+ return err
+ }
+ if _, err := sim.WaitTillHealthy(ctx, 2); err != nil {
+ return err
}
- return allSuccess, nil
- }
- //for each tick, run the checks on all nodes
- timingTicker := time.NewTicker(5 * time.Second)
- defer timingTicker.Stop()
- go func() {
- for range timingTicker.C {
- for i := 0; i < len(ids); i++ {
- log.Trace(fmt.Sprintf("triggering step %d, id %s", i, ids[i]))
- trigger <- ids[i]
+ // File retrieval check is repeated until all uploaded files are retrieved from all nodes
+ // or until the timeout is reached.
+ allSuccess := false
+ for !allSuccess {
+ for _, id := range nodeIDs {
+ //for each expected chunk, check if it is in the local store
+ localChunks := conf.idToChunksMap[id]
+ localSuccess := true
+ for _, ch := range localChunks {
+ //get the real chunk by the index in the index array
+ chunk := conf.hashes[ch]
+ log.Trace(fmt.Sprintf("node has chunk: %s:", chunk))
+ //check if the expected chunk is indeed in the localstore
+ var err error
+ //check on the node's FileStore (netstore)
+ item, ok := sim.NodeItem(id, bucketKeyFileStore)
+ if !ok {
+ return fmt.Errorf("No registry")
+ }
+ fileStore := item.(*storage.FileStore)
+ //check all chunks
+ for i, hash := range conf.hashes {
+ reader, _ := fileStore.Retrieve(context.TODO(), hash)
+ //check that we can read the file size and that it corresponds to the generated file size
+ if s, err := reader.Size(ctx, nil); err != nil || s != int64(len(randomFiles[i])) {
+ allSuccess = false
+ log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id)
+ } else {
+ log.Debug(fmt.Sprintf("File with root hash %x successfully retrieved", hash))
+ }
+ }
+ if err != nil {
+ log.Warn(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id))
+ localSuccess = false
+ } else {
+ log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", chunk, id))
+ }
+ }
+ allSuccess = localSuccess
}
}
- }()
-
- log.Info("Starting simulation run...")
-
- timeout := MaxTimeout * time.Second
- ctx, cancel := context.WithTimeout(context.Background(), timeout)
- defer cancel()
-
- //run the simulation
- result := simulations.NewSimulation(net).Run(ctx, &simulations.Step{
- Action: action,
- Trigger: trigger,
- Expect: &simulations.Expectation{
- Nodes: ids,
- Check: check,
- },
+ if !allSuccess {
+ return fmt.Errorf("Not all chunks succeeded!")
+ }
+ return nil
})
if result.Error != nil {
@@ -743,53 +388,3 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
return nil
}
-
-//upload generated files to nodes
-//every node gets one file uploaded
-func uploadFilesToNodes(nodes []*simulations.Node) ([]storage.Address, []string, error) {
- nodeCnt := len(nodes)
- log.Debug(fmt.Sprintf("Uploading %d files to nodes", nodeCnt))
- //array holding generated files
- rfiles := make([]string, nodeCnt)
- //array holding the root hashes of the files
- rootAddrs := make([]storage.Address, nodeCnt)
-
- var err error
- //for every node, generate a file and upload
- for i, n := range nodes {
- id := n.ID()
- fileStore := registries[id].fileStore
- //generate a file
- rfiles[i], err = generateRandomFile()
- if err != nil {
- return nil, nil, err
- }
- //store it (upload it) on the FileStore
- ctx := context.TODO()
- rk, wait, err := fileStore.Store(ctx, strings.NewReader(rfiles[i]), int64(len(rfiles[i])), false)
- log.Debug("Uploaded random string file to node")
- if err != nil {
- return nil, nil, err
- }
- err = wait(ctx)
- if err != nil {
- return nil, nil, err
- }
- rootAddrs[i] = rk
- }
- return rootAddrs, rfiles, nil
-}
-
-//generate a random file (string)
-func generateRandomFile() (string, error) {
- //generate a random file size between minFileSize and maxFileSize
- fileSize := rand.Intn(maxFileSize-minFileSize) + minFileSize
- log.Debug(fmt.Sprintf("Generated file with filesize %d kB", fileSize))
- b := make([]byte, fileSize*1024)
- _, err := crand.Read(b)
- if err != nil {
- log.Error("Error generating random file.", "err", err)
- return "", err
- }
- return string(b), nil
-}