swarm: fix relationship between spans in open tracing (#19236)
* swarm/network: propagate span with ctx * swarm/network: try to stop stream.send.request spans on time * swarm/storage: add chunk ref as a log to netstore.fetcher span
This commit is contained in:
parent
ceeb047e69
commit
2cfe0bed9f
@ -18,12 +18,15 @@ package network
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"fmt"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/ethereum/go-ethereum/log"
|
"github.com/ethereum/go-ethereum/log"
|
||||||
"github.com/ethereum/go-ethereum/p2p/enode"
|
"github.com/ethereum/go-ethereum/p2p/enode"
|
||||||
"github.com/ethereum/go-ethereum/swarm/storage"
|
"github.com/ethereum/go-ethereum/swarm/storage"
|
||||||
|
"github.com/ethereum/go-ethereum/swarm/tracing"
|
||||||
|
"github.com/opentracing/opentracing-go"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
@ -318,6 +321,14 @@ func (f *Fetcher) doRequest(gone chan *enode.ID, peersToSkip *sync.Map, sources
|
|||||||
gone <- sourceID
|
gone <- sourceID
|
||||||
case <-f.ctx.Done():
|
case <-f.ctx.Done():
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// finish the request span
|
||||||
|
spanId := fmt.Sprintf("stream.send.request.%v.%v", *sourceID, req.Addr)
|
||||||
|
span := tracing.ShiftSpanByKey(spanId)
|
||||||
|
|
||||||
|
if span != nil {
|
||||||
|
defer span.(opentracing.Span).Finish()
|
||||||
|
}
|
||||||
}()
|
}()
|
||||||
return sources, nil
|
return sources, nil
|
||||||
}
|
}
|
||||||
|
@ -160,7 +160,7 @@ func (p *Peer) Deliver(ctx context.Context, chunk storage.Chunk, priority uint8,
|
|||||||
// SendPriority sends message to the peer using the outgoing priority queue
|
// SendPriority sends message to the peer using the outgoing priority queue
|
||||||
func (p *Peer) SendPriority(ctx context.Context, msg interface{}, priority uint8) 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())
|
defer metrics.GetOrRegisterResettingTimer(fmt.Sprintf("peer.sendpriority_t.%d", priority), nil).UpdateSince(time.Now())
|
||||||
tracing.StartSaveSpan(ctx)
|
ctx = tracing.StartSaveSpan(ctx)
|
||||||
metrics.GetOrRegisterCounter(fmt.Sprintf("peer.sendpriority.%d", priority), nil).Inc(1)
|
metrics.GetOrRegisterCounter(fmt.Sprintf("peer.sendpriority.%d", priority), nil).Inc(1)
|
||||||
wmsg := WrappedPriorityMsg{
|
wmsg := WrappedPriorityMsg{
|
||||||
Context: ctx,
|
Context: ctx,
|
||||||
|
@ -28,6 +28,7 @@ import (
|
|||||||
"github.com/ethereum/go-ethereum/swarm/log"
|
"github.com/ethereum/go-ethereum/swarm/log"
|
||||||
"github.com/ethereum/go-ethereum/swarm/spancontext"
|
"github.com/ethereum/go-ethereum/swarm/spancontext"
|
||||||
"github.com/opentracing/opentracing-go"
|
"github.com/opentracing/opentracing-go"
|
||||||
|
olog "github.com/opentracing/opentracing-go/log"
|
||||||
|
|
||||||
lru "github.com/hashicorp/golang-lru"
|
lru "github.com/hashicorp/golang-lru"
|
||||||
)
|
)
|
||||||
@ -215,6 +216,8 @@ func (n *NetStore) getOrCreateFetcher(ctx context.Context, ref Address) *fetcher
|
|||||||
cctx,
|
cctx,
|
||||||
"netstore.fetcher",
|
"netstore.fetcher",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
sp.LogFields(olog.String("ref", ref.String()))
|
||||||
fetcher := newFetcher(sp, ref, n.NewNetFetcherFunc(cctx, ref, peers), destroy, peers, n.closeC)
|
fetcher := newFetcher(sp, ref, n.NewNetFetcherFunc(cctx, ref, peers), destroy, peers, n.closeC)
|
||||||
n.fetchers.Add(key, fetcher)
|
n.fetchers.Add(key, fetcher)
|
||||||
|
|
||||||
|
Loading…
Reference in New Issue
Block a user