From b807d23991e43312d3a4062377c82382ccbe3d0f Mon Sep 17 00:00:00 2001 From: Fabio Barone Date: Fri, 14 Sep 2018 14:52:57 -0500 Subject: [PATCH] swarm/network/stream: introduced visualized snapshot sync test --- p2p/simulations/events.go | 3 + swarm/network/stream/messages.go | 1 + swarm/network/stream/peer.go | 5 + swarm/network/stream/snapshot_sync_test.go | 112 +++++++++---- .../visualized_snapshot_sync_sim_test.go | 158 ++++++++++++++++++ swarm/tracing/tracing.go | 8 + 6 files changed, 250 insertions(+), 37 deletions(-) create mode 100644 swarm/network/stream/visualized_snapshot_sync_sim_test.go diff --git a/p2p/simulations/events.go b/p2p/simulations/events.go index f17958c689..c49a72b596 100644 --- a/p2p/simulations/events.go +++ b/p2p/simulations/events.go @@ -58,6 +58,9 @@ type Event struct { // Msg is set if the type is EventTypeMsg Msg *Msg `json:"msg,omitempty"` + + // + Data interface{} `json:"data"` } // NewEvent creates a new event for the given object which should be either a diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index 74c785d587..ff54325242 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -267,6 +267,7 @@ func (p *Peer) handleOfferedHashesMsg(ctx context.Context, req *OfferedHashesMsg return nil } + sp.SetTag("hash", fmt.Sprintf("%064x", want.Bytes()[:])) msg := &WantedHashesMsg{ Stream: req.Stream, Want: want.Bytes(), diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index ef6bbdf70d..41627580ca 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -133,6 +133,7 @@ func (p *Peer) Deliver(ctx context.Context, chunk storage.Chunk, priority uint8) ctx, sp = spancontext.StartSpan( ctx, "send.chunk.delivery") + sp.SetTag("hash", chunk.Address().String()) defer sp.Finish() msg := &ChunkDeliveryMsg{ @@ -167,6 +168,10 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error { defer sp.Finish() hashes, from, to, proof, err := s.SetNextBatch(f, t) + for i := 0; i < len(hashes); i += HashSize { + hash := hashes[i : i+HashSize] + sp.SetTag("hash", fmt.Sprintf("%064x", hash[:])) + } if err != nil { return err } diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index d93afce1b1..3d59f6b8a2 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/swarm/network/stream/snapshot_sync_test.go @@ -30,7 +30,9 @@ import ( "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/enode" + "github.com/ethereum/go-ethereum/p2p/simulations" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/network/simulation" @@ -50,6 +52,17 @@ type synctestConfig struct { addrToIDMap map[string]enode.ID } +const ( + // EventTypeNode is the type of event emitted when a node is either + // created, started or stopped + EventTypeChunkCreated simulations.EventType = "chunkCreated" + EventTypeChunkOffered simulations.EventType = "chunkOffered" + EventTypeChunkWanted simulations.EventType = "chunkWanted" + EventTypeChunkDelivered simulations.EventType = "chunkDelivered" + EventTypeChunkArrived simulations.EventType = "chunkArrived" + EventTypeSimTerminated simulations.EventType = "simTerminated" +) + // Tests in this file should not request chunks from peers. // This function will panic indicating that there is a problem if request has been made. func dummyRequestFromPeers(_ context.Context, req *network.Request) (*enode.ID, chan struct{}, error) { @@ -131,41 +144,47 @@ func TestSyncingViaDirectSubscribe(t *testing.T) { } } -func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) { - sim := simulation.New(map[string]simulation.ServiceFunc{ - "streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) { - n := ctx.Config.Node() - addr := network.NewAddr(n) - store, datadir, err := createTestLocalStorageForID(n.ID(), addr) - if err != nil { - return nil, nil, err - } - bucket.Store(bucketKeyStore, store) - localStore := store.(*storage.LocalStore) - netStore, err := storage.NewNetStore(localStore, nil) - if err != nil { - return nil, nil, err - } - kad := network.NewKademlia(addr.Over(), network.NewKadParams()) - delivery := NewDelivery(kad, netStore) - netStore.NewNetFetcherFunc = network.NewFetcherFactory(dummyRequestFromPeers, true).New +var simServiceMap = map[string]simulation.ServiceFunc{ + "streamer": streamerFunc, +} - r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{ - DoSync: true, - SyncUpdateDelay: 3 * time.Second, - }) - bucket.Store(bucketKeyRegistry, r) +func streamerFunc(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) { - cleanup = func() { - os.RemoveAll(datadir) - netStore.Close() - r.Close() - } + n := ctx.Config.Node() + addr := network.NewAddr(n) + store, datadir, err := createTestLocalStorageForID(n.ID(), addr) + if err != nil { + return nil, nil, err + } + bucket.Store(bucketKeyStore, store) + localStore := store.(*storage.LocalStore) + netStore, err := storage.NewNetStore(localStore, nil) + if err != nil { + return nil, nil, err + } + kad := network.NewKademlia(addr.Over(), network.NewKadParams()) + delivery := NewDelivery(kad, netStore) + netStore.NewNetFetcherFunc = network.NewFetcherFactory(dummyRequestFromPeers, true).New - return r, cleanup, nil - - }, + r := NewRegistry(addr, delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{ + DoSync: true, + SyncUpdateDelay: 3 * time.Second, }) + + bucket.Store(bucketKeyRegistry, r) + + cleanup = func() { + os.RemoveAll(datadir) + netStore.Close() + r.Close() + } + + return r, cleanup, nil + +} + +func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) { + sim := simulation.New(simServiceMap) defer sim.Close() log.Info("Initializing test config") @@ -204,7 +223,17 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) { } }() - result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error { + result := runSim(conf, ctx, sim, chunkCount) + + if result.Error != nil { + t.Fatal(result.Error) + } + log.Info("Simulation ended") +} + +func runSim(conf *synctestConfig, ctx context.Context, sim *simulation.Simulation, chunkCount int) simulation.Result { + + return 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 @@ -229,6 +258,14 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) { if err != nil { return err } + for _, h := range hashes { + evt := &simulations.Event{ + Type: EventTypeChunkCreated, + Node: sim.Net.GetNode(node.ID), + Data: fmt.Sprintf("%s", h), + } + sim.Net.Events().Send(evt) + } conf.hashes = append(conf.hashes, hashes...) mapKeysToNodes(conf) @@ -281,6 +318,12 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) { // Do not get crazy with logging the warn message time.Sleep(500 * time.Millisecond) } else { + evt := &simulations.Event{ + Type: EventTypeChunkArrived, + Node: sim.Net.GetNode(id), + Data: fmt.Sprintf("%s", chunk), + } + sim.Net.Events().Send(evt) log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", chunk, id)) } } @@ -295,11 +338,6 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) { } return nil }) - - if result.Error != nil { - t.Fatal(result.Error) - } - log.Info("Simulation ended") } /* diff --git a/swarm/network/stream/visualized_snapshot_sync_sim_test.go b/swarm/network/stream/visualized_snapshot_sync_sim_test.go new file mode 100644 index 0000000000..5fa80ed6c1 --- /dev/null +++ b/swarm/network/stream/visualized_snapshot_sync_sim_test.go @@ -0,0 +1,158 @@ +// +build withserver + +package stream + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/simulations" + "github.com/ethereum/go-ethereum/swarm/log" + "github.com/ethereum/go-ethereum/swarm/network/simulation" + "github.com/ethereum/go-ethereum/swarm/storage" + // "github.com/ethereum/go-ethereum/swarm/tracing" +) + +func init() { + /* + var flagSet *flag.FlagSet + tracing.Enabled = true + tracing.StandaloneSetup() + fakeApp := cli.NewApp() + flags := []cli.Flag{ + tracing.TracingEndpointFlag, + tracing.TracingSvcFlag, + } + fakeApp.Flags = append(fakeApp.Flags, flags...) + fakeApp.Before = func(ctx *cli.Context) error { + tracing.Setup(ctx) + return nil + } + fakeApp.Run([]string{"-tracing.endpoint", tracing.TracingEndpointFlag.Value, "-tracing.svc", tracing.TracingSvcFlag.Value}) + + //flagSet = flag.NewFlagSet("traceFlags", 0) + //tracing.Setup(cli.NewContext(fakeApp, flagSet, nil)) + */ +} + +func TestSnapshotSyncWithServer(t *testing.T) { + + nodeCount := *nodes + chunkCount := *chunks + + if nodeCount == 0 || chunkCount == 0 { + nodeCount = 32 + chunkCount = 1 + } + + sim := simulation.New(simServiceMap).WithServer(":8888") + defer sim.Close() + + log.Info("Initializing test config") + + 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) + + err := sim.UploadSnapshot(fmt.Sprintf("testing/snapshot_%d.json", nodeCount)) + if err != nil { + panic(err) + } + + ctx, cancelSimRun := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancelSimRun() + + if _, err := sim.WaitTillHealthy(ctx, 2); err != nil { + panic(err) + } + + disconnections := sim.PeerEvents( + context.Background(), + sim.NodeIDs(), + simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeDrop), + ) + + go func() { + for d := range disconnections { + log.Error("peer drop", "node", d.NodeID, "peer", d.Event.Peer) + panic("unexpected disconnect") + cancelSimRun() + } + }() + + //sim.PeerEvents( + offeredHashesFilter := simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeMsgRecv).Protocol("stream").MsgCode(1) + wantedFilter := simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeMsgRecv).Protocol("stream").MsgCode(2) + deliveryFilter := simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeMsgRecv).Protocol("stream").MsgCode(6) + eventC := sim.PeerEvents(ctx, sim.UpNodeIDs(), offeredHashesFilter, wantedFilter, deliveryFilter) + + quit := make(chan struct{}) + + go func() { + for e := range eventC { + select { + case <-quit: + fmt.Println("quitting event loop") + return + default: + } + if e.Error != nil { + t.Fatal(e.Error) + } + if *e.Event.MsgCode == uint64(1) { + evt := &simulations.Event{ + Type: EventTypeChunkOffered, + Node: sim.Net.GetNode(e.NodeID), + //Data: fmt.Sprintf("%s", h), + } + sim.Net.Events().Send(evt) + } else if *e.Event.MsgCode == uint64(2) { + evt := &simulations.Event{ + Type: EventTypeChunkWanted, + Node: sim.Net.GetNode(e.NodeID), + //Data: fmt.Sprintf("%s", h), + } + sim.Net.Events().Send(evt) + } else if *e.Event.MsgCode == uint64(6) { + evt := &simulations.Event{ + Type: EventTypeChunkDelivered, + Node: sim.Net.GetNode(e.NodeID), + //Data: fmt.Sprintf("%s", h), + } + sim.Net.Events().Send(evt) + } + } + }() + result := runSim(conf, ctx, sim, chunkCount) + + evt := &simulations.Event{ + Type: EventTypeSimTerminated, + } + sim.Net.Events().Send(evt) + + if result.Error != nil { + panic(result.Error) + } + close(quit) + log.Info("Simulation ended") +} + +/* +func decodeMsg(code int) error { + val, ok := Spec.NewMsg(code) + if !ok { + return errorf("invalid msg code", "%v", msg.Code) + } + if err := rlp.DecodeBytes(wmsg.Payload, val); err != nil { + return errorf(ErrDecode, "<= %v: %v", msg, err) + } +} +*/ diff --git a/swarm/tracing/tracing.go b/swarm/tracing/tracing.go index b84cfb3102..ef393098a4 100644 --- a/swarm/tracing/tracing.go +++ b/swarm/tracing/tracing.go @@ -67,6 +67,14 @@ func Setup(ctx *cli.Context) { } } +func StandaloneSetup() { + if Enabled { + log.Info("Enabling opentracing") + + Closer = initTracer(TracingEndpointFlag.Value, TracingSvcFlag.Value) + } +} + func initTracer(endpoint, svc string) (closer io.Closer) { // Sample configuration for testing. Use constant sampling to sample every trace // and enable LogSpan to log every span via configured Logger.