swarm/network: Put span ids for sendpriority in context value

swarm/network/stream/delivery.go

@@ -143,7 +143,7 @@ func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req *
 	var osp opentracing.Span
 	ctx, osp = spancontext.StartSpan(
 		ctx,
-		"retrieve.request")
+		"stream.handle.retrieve")
 
 	s, err := sp.getServer(NewStream(swarmChunkServerStreamName, "", true))
 	if err != nil {
@@ -207,21 +207,17 @@ type ChunkDeliveryMsgRetrieval ChunkDeliveryMsg
 //defines a chunk delivery for syncing (without accounting)
 type ChunkDeliveryMsgSyncing ChunkDeliveryMsg
 
-// TODO: Fix context SNAFU
+// chunk delivery msg is response to retrieverequest msg
 func (d *Delivery) handleChunkDeliveryMsg(ctx context.Context, sp *Peer, req *ChunkDeliveryMsg) error {
-	//	var osp opentracing.Span
-	//	ctx, osp = spancontext.StartSpan(
-	//		ctx,
-	//		"chunk.delivery")
-
-	spanId := fmt.Sprintf("request.%v.%v", sp.ID(), req.Addr)
-	span, spanOk := sp.spans.Load(spanId)
-	sp.spans.Delete(spanId)
 
 	processReceivedChunksCount.Inc(1)
 
+	// retrieve the span for the originating retrieverequest
+	spanId := fmt.Sprintf("stream.send.request.%v.%v", sp.ID(), req.Addr)
+	span, spanOk := sp.spans.Load(spanId)
+	sp.spans.Delete(spanId)
+
 	go func() {
-		//defer osp.Finish()
 		if spanOk {
 			defer span.(opentracing.Span).Finish()
 		}
@@ -240,7 +236,9 @@ func (d *Delivery) handleChunkDeliveryMsg(ctx context.Context, sp *Peer, req *Ch
 	return nil
 }
 
-// RequestFromPeers sends a chunk retrieve request to
+// RequestFromPeers sends a chunk retrieve request to a peer
+// The most eligible peer that hasn't already been sent to is chosen
+// TODO: define "eligible"
 func (d *Delivery) RequestFromPeers(ctx context.Context, req *network.Request) (*enode.ID, chan struct{}, error) {
 	requestFromPeersCount.Inc(1)
 	var sp *Peer
@@ -275,11 +273,15 @@ func (d *Delivery) RequestFromPeers(ctx context.Context, req *network.Request) (
 		}
 	}
 
+	// setting this value in the context creates a new span that can persist across the sendpriority queue and the network roundtrip
+	// this span will finish only when delivery is handled (or times out)
+	ctx = context.WithValue(ctx, "stream_send_tag", "stream.send.request")
+	ctx = context.WithValue(ctx, "stream_send_meta", fmt.Sprintf("%v.%v", sp.ID(), req.Addr))
 	err := sp.SendPriority(ctx, &RetrieveRequestMsg{
 		Addr:      req.Addr,
 		SkipCheck: req.SkipCheck,
 		HopCount:  req.HopCount,
-	}, Top, fmt.Sprintf("request.%v.%v", sp.ID(), req.Addr))
+	}, Top)
 	if err != nil {
 		return nil, nil, err
 	}

swarm/network/stream/messages.go

@@ -300,7 +300,7 @@ func (p *Peer) handleOfferedHashesMsg(ctx context.Context, req *OfferedHashesMsg
 			return
 		}
 		log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "from", msg.From, "to", msg.To)
-		err := p.SendPriority(ctx, msg, c.priority, "")
+		err := p.SendPriority(ctx, msg, c.priority)
 		if err != nil {
 			log.Warn("SendPriority error", "err", err)
 		}

swarm/network/stream/peer.go

@@ -158,20 +158,26 @@ func (p *Peer) Deliver(ctx context.Context, chunk storage.Chunk, priority uint8,
 		spanName += ".retrieval"
 	}
 
-	//return p.SendPriority(ctx, msg, priority, spanName)
+	ctx = context.WithValue(ctx, "stream_send_tag", nil)
-	return p.SendPriority(ctx, msg, priority, "")
+	return p.SendPriority(ctx, msg, priority)
 }
 
 // SendPriority sends message to the peer using the outgoing priority queue
-func (p *Peer) SendPriority(ctx context.Context, msg interface{}, priority uint8, traceId string) error {
+func (p *Peer) SendPriority(ctx context.Context, msg interface{}, priority uint8) error {
 	defer metrics.GetOrRegisterResettingTimer(fmt.Sprintf("peer.sendpriority_t.%d", priority), nil).UpdateSince(time.Now())
 	metrics.GetOrRegisterCounter(fmt.Sprintf("peer.sendpriority.%d", priority), nil).Inc(1)
-	if traceId != "" {
+	traceId := ctx.Value("stream_send_tag")
+	if traceId != nil {
+		traceStr := traceId.(string)
 		var sp opentracing.Span
 		ctx, sp = spancontext.StartSpan(
 			ctx,
-			traceId,
+			traceStr,
 		)
+		traceMeta := ctx.Value("stream_send_meta")
+		if traceMeta != nil {
+			traceStr = traceStr + "." + traceMeta.(string)
+		}
 		p.spans.Store(traceId, sp)
 	}
 	wmsg := WrappedPriorityMsg{
@@ -217,7 +223,8 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error {
 		Stream:        s.stream,
 	}
 	log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "len", len(hashes), "from", from, "to", to)
-	return p.SendPriority(ctx, msg, s.priority, "send.offered.hashes")
+	ctx = context.WithValue(ctx, "stream_send_tag", "send.offered.hashes")
+	return p.SendPriority(ctx, msg, s.priority)
 }
 
 func (p *Peer) getServer(s Stream) (*server, error) {

swarm/network/stream/stream.go

@@ -359,7 +359,7 @@ func (r *Registry) Subscribe(peerId enode.ID, s Stream, h *Range, priority uint8
 	}
 	log.Debug("Subscribe ", "peer", peerId, "stream", s, "history", h)
 
-	return peer.SendPriority(context.TODO(), msg, priority, "")
+	return peer.SendPriority(context.TODO(), msg, priority)
 }
 
 func (r *Registry) Unsubscribe(peerId enode.ID, s Stream) error {
@@ -730,7 +730,7 @@ func (c *client) batchDone(p *Peer, req *OfferedHashesMsg, hashes []byte) error
 			return err
 		}
 
-		if err := p.SendPriority(context.TODO(), tp, c.priority, ""); err != nil {
+		if err := p.SendPriority(context.TODO(), tp, c.priority); err != nil {
 			return err
 		}
 		if c.to > 0 && tp.Takeover.End >= c.to {