swarm/network/stream: introduced visualized snapshot sync test

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

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(),

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
 	}

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) {
+var simServiceMap = map[string]simulation.ServiceFunc{
-	sim := simulation.New(map[string]simulation.ServiceFunc{
+	"streamer": streamerFunc,
-		"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
 
-			r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
+func streamerFunc(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
-				DoSync:          true,
-				SyncUpdateDelay: 3 * time.Second,
-			})
-			bucket.Store(bucketKeyRegistry, r)
 
-			cleanup = func() {
+	n := ctx.Config.Node()
-				os.RemoveAll(datadir)
+	addr := network.NewAddr(n)
-				netStore.Close()
+	store, datadir, err := createTestLocalStorageForID(n.ID(), addr)
-				r.Close()
+	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")
 }
 
 /*

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)
+	}
+}
+*/

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.