Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add basic tracing for responses #291

Merged
merged 4 commits into from
Dec 11, 2021
Merged

feat: add basic tracing for responses #291

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6b1439e
feat: add basic tracing for responses
rvagg Dec 2, 2021
80fcd3a
fix: add assertions for response completion
hannahhoward Dec 9, 2021
1c3d63a
fix: improve test synchronization/drainage
rvagg Dec 10, 2021
eff0a0a
fix(responsemanager): remove uninformative span flags
hannahhoward Dec 11, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
289 changes: 213 additions & 76 deletions impl/graphsync_test.go
View file
Open in desktop

Large diffs are not rendered by default.

17 changes: 17 additions & 0 deletions message/message.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -375,6 +375,15 @@ func (gsr GraphSyncRequest) Extension(name graphsync.ExtensionName) ([]byte, boo
return val, true
}

// ExtensionNames returns the names of the extensions included in this request
func (gsr GraphSyncRequest) ExtensionNames() []string {
var extNames []string
for ext := range gsr.extensions {
extNames = append(extNames, ext)
}
return extNames
}

// IsCancel returns true if this particular request is being cancelled
func (gsr GraphSyncRequest) IsCancel() bool { return gsr.isCancel }

Expand All @@ -398,7 +407,15 @@ func (gsr GraphSyncResponse) Extension(name graphsync.ExtensionName) ([]byte, bo
return nil, false
}
return val, true
}

// ExtensionNames returns the names of the extensions included in this request
func (gsr GraphSyncResponse) ExtensionNames() []string {
var extNames []string
for ext := range gsr.extensions {
extNames = append(extNames, ext)
}
return extNames
}

// ReplaceExtensions merges the extensions given extensions into the request to create a new request,
Expand Down
18 changes: 10 additions & 8 deletions requestmanager/executor/executor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,14 +82,16 @@ func (e *Executor) ExecuteTask(ctx context.Context, pid peer.ID, task *peertask.

log.Debugw("beginning request execution", "id", requestTask.Request.ID(), "peer", pid.String(), "root_cid", requestTask.Request.Root().String())
err := e.traverse(requestTask)
span.RecordError(err)
if err != nil && !ipldutil.IsContextCancelErr(err) {
e.manager.SendRequest(requestTask.P, gsmsg.CancelRequest(requestTask.Request.ID()))
if !isPausedErr(err) {
span.SetStatus(codes.Error, err.Error())
select {
case <-requestTask.Ctx.Done():
case requestTask.InProgressErr <- err:
if err != nil {
span.RecordError(err)
if !ipldutil.IsContextCancelErr(err) {
e.manager.SendRequest(requestTask.P, gsmsg.CancelRequest(requestTask.Request.ID()))
if !isPausedErr(err) {
span.SetStatus(codes.Error, err.Error())
select {
case <-requestTask.Ctx.Done():
case requestTask.InProgressErr <- err:
}
}
}
}
Expand Down
4 changes: 3 additions & 1 deletion responsemanager/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"github.com/ipfs/go-peertaskqueue/peertask"
ipld "github.com/ipld/go-ipld-prime"
"github.com/libp2p/go-libp2p-core/peer"
"go.opentelemetry.io/otel/trace"

"github.com/ipfs/go-graphsync"
"github.com/ipfs/go-graphsync/ipldutil"
Expand All @@ -30,6 +31,7 @@ var log = logging.Logger("graphsync")

type inProgressResponseStatus struct {
ctx context.Context
span trace.Span
cancelFn func()
request gsmsg.GraphSyncRequest
loader ipld.BlockReadOpener
Expand Down Expand Up @@ -197,7 +199,7 @@ func (rm *ResponseManager) CancelResponse(p peer.ID, requestID graphsync.Request
}
}

// this is a test utility method to force all messages to get processed
// Synchronize is a utility method that blocks until all current messages are processed
func (rm *ResponseManager) synchronize() {
sync := make(chan error)
rm.send(&synchronizeMessage{sync}, nil)
Expand Down
8 changes: 4 additions & 4 deletions responsemanager/messages.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,10 @@ type processRequestMessage struct {
requests []gsmsg.GraphSyncRequest
}

func (prm *processRequestMessage) handle(rm *ResponseManager) {
rm.processRequests(prm.p, prm.requests)
}

type pauseRequestMessage struct {
p peer.ID
requestID graphsync.RequestID
Expand Down Expand Up @@ -111,10 +115,6 @@ func (str *startTaskRequest) handle(rm *ResponseManager) {
}
}

func (prm *processRequestMessage) handle(rm *ResponseManager) {
rm.processRequests(prm.p, prm.requests)
}

type peerStateMessage struct {
p peer.ID
peerStatsChan chan<- peerstate.PeerState
Expand Down
13 changes: 13 additions & 0 deletions responsemanager/queryexecutor/queryexecutor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,9 @@ import (
ipld "github.com/ipld/go-ipld-prime"
"github.com/ipld/go-ipld-prime/traversal"
"github.com/libp2p/go-libp2p-core/peer"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/trace"

"github.com/ipfs/go-graphsync"
"github.com/ipfs/go-graphsync/ipldutil"
Expand Down Expand Up @@ -38,6 +41,7 @@ type ResponseTask struct {
Empty bool
Subscriber *notifications.TopicDataSubscriber
Ctx context.Context
Span trace.Span
Request gsmsg.GraphSyncRequest
Loader ipld.BlockReadOpener
Traverser ipldutil.Traverser
Expand Down Expand Up @@ -97,8 +101,17 @@ func (qe *QueryExecutor) ExecuteTask(ctx context.Context, pid peer.ID, task *pee
return false
}

_, span := otel.Tracer("graphsync").Start(trace.ContextWithSpan(qe.ctx, rt.Span), "executeTask")
defer span.End()

log.Debugw("beginning response execution", "id", rt.Request.ID(), "peer", pid.String(), "root_cid", rt.Request.Root().String())
err := qe.executeQuery(pid, rt)
if err != nil {
span.RecordError(err)
if _, isPaused := err.(hooks.ErrPaused); !isPaused {
span.SetStatus(codes.Error, err.Error())
}
}
qe.manager.FinishTask(task, err)
log.Debugw("finishing response execution", "id", rt.Request.ID(), "peer", pid.String(), "root_cid", rt.Request.Root().String())
return false
Expand Down
37 changes: 36 additions & 1 deletion responsemanager/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,10 @@ import (
"github.com/ipfs/go-peertaskqueue/peertask"
"github.com/ipfs/go-peertaskqueue/peertracker"
"github.com/libp2p/go-libp2p-core/peer"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/trace"

"github.com/ipfs/go-graphsync"
"github.com/ipfs/go-graphsync/ipldutil"
Expand Down Expand Up @@ -50,6 +54,7 @@ func (rm *ResponseManager) terminateRequest(key responseKey) {
rm.connManager.Unprotect(key.p, key.requestID.Tag())
delete(rm.inProgressResponses, key)
ipr.cancelFn()
ipr.span.End()
}

func (rm *ResponseManager) processUpdate(key responseKey, update gsmsg.GraphSyncRequest) {
Expand All @@ -58,6 +63,13 @@ func (rm *ResponseManager) processUpdate(key responseKey, update gsmsg.GraphSync
log.Warnf("received update for non existent request, peer %s, request ID %d", key.p.Pretty(), key.requestID)
return
}

_, span := otel.Tracer("graphsync").Start(trace.ContextWithSpan(rm.ctx, response.span), "processUpdate", trace.WithAttributes(
attribute.Int("id", int(update.ID())),
attribute.StringSlice("extensions", update.ExtensionNames()),
))
defer span.End()

if response.state != graphsync.Paused {
response.updates = append(response.updates, update)
select {
Expand All @@ -79,11 +91,15 @@ func (rm *ResponseManager) processUpdate(key responseKey, update gsmsg.GraphSync
})
if result.Err != nil {
response.state = graphsync.CompletingSend
response.span.RecordError(result.Err)
response.span.SetStatus(codes.Error, result.Err.Error())
return
}
if result.Unpause {
err := rm.unpauseRequest(key.p, key.requestID)
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, result.Err.Error())
log.Warnf("error unpausing request: %s", err.Error())
}
}
Expand Down Expand Up @@ -119,6 +135,13 @@ func (rm *ResponseManager) abortRequest(p peer.ID, requestID graphsync.RequestID
return errors.New("could not find request")
}

_, span := otel.Tracer("graphsync").Start(trace.ContextWithSpan(rm.ctx, response.span), "abortRequest")
defer span.End()
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
response.span.RecordError(err)
response.span.SetStatus(codes.Error, err.Error())
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I dunno what good span error conventions are but should we record this twice at two different levels?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, I don't know and wanted to discuss this with you. This particular one is a bit tricky, others are a little more clear I think that "this error belongs to the response" but here there's span for something called "abort" and as a user I'd expect to see what that abort was about on that span, but the error is also responsible for closing the request.
I could go either way on this, and threw up my hands and said why not both? 🌮

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

all good by me :)


if response.state != graphsync.Running {
_ = rm.responseAssembler.Transaction(p, requestID, func(rb responseassembler.ResponseBuilder) error {
if ipldutil.IsContextCancelErr(err) {
Expand Down Expand Up @@ -155,9 +178,15 @@ func (rm *ResponseManager) processRequests(p peer.ID, requests []gsmsg.GraphSync
rm.processUpdate(key, request)
continue
}
ctx, responseSpan := otel.Tracer("graphsync").Start(rm.ctx, "response", trace.WithAttributes(
attribute.Int("id", int(request.ID())),
attribute.Int("priority", int(request.Priority())),
attribute.String("root", request.Root().String()),
attribute.StringSlice("extensions", request.ExtensionNames()),
))
rm.connManager.Protect(p, request.ID().Tag())
rm.requestQueuedHooks.ProcessRequestQueuedHooks(p, request)
ctx, cancelFn := context.WithCancel(rm.ctx)
ctx, cancelFn := context.WithCancel(ctx)
sub := notifications.NewTopicDataSubscriber(&subscriber{
p: key.p,
request: request,
Expand All @@ -176,6 +205,7 @@ func (rm *ResponseManager) processRequests(p peer.ID, requests []gsmsg.GraphSync
rm.inProgressResponses[key] =
&inProgressResponseStatus{
ctx: ctx,
span: responseSpan,
cancelFn: cancelFn,
subscriber: sub,
request: request,
Expand Down Expand Up @@ -204,6 +234,8 @@ func (rm *ResponseManager) taskDataForKey(key responseKey) queryexecutor.Respons
loader, traverser, isPaused, err := (&queryPreparer{rm.requestHooks, rm.responseAssembler, rm.linkSystem, rm.maxLinksPerRequest}).prepareQuery(response.ctx, key.p, response.request, response.signals, response.subscriber)
if err != nil {
response.state = graphsync.CompletingSend
response.span.RecordError(err)
response.span.SetStatus(codes.Error, err.Error())
return queryexecutor.ResponseTask{Empty: true}
}
response.loader = loader
Expand All @@ -216,6 +248,7 @@ func (rm *ResponseManager) taskDataForKey(key responseKey) queryexecutor.Respons
response.state = graphsync.Running
return queryexecutor.ResponseTask{
Ctx: response.ctx,
Span: response.span,
Empty: false,
Subscriber: response.subscriber,
Request: response.request,
Expand Down Expand Up @@ -249,6 +282,8 @@ func (rm *ResponseManager) finishTask(task *peertask.Task, err error) {
log.Infow("graphsync response processing complete (messages stil sending)", "request id", key.requestID, "peer", key.p, "total time", time.Since(response.startTime))

if err != nil {
response.span.RecordError(err)
response.span.SetStatus(codes.Error, err.Error())
log.Infof("response failed: %w", err)
}

Expand Down
10 changes: 5 additions & 5 deletions testutil/tracing.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,13 +77,13 @@ func (c Collector) tracesToString(trace string, spans tracetest.SpanStubs, match
// identified by its trace string as described in TracesToStrings. Note that
// this string can also be a partial of a complete trace, e.g. just `"foo(0)"`
// without any children to fetch the parent span.
func (c Collector) FindSpanByTraceString(trace string) tracetest.SpanStub {
var found tracetest.SpanStub
func (c Collector) FindSpanByTraceString(trace string) *tracetest.SpanStub {
var found *tracetest.SpanStub
c.tracesToString("", c.FindParentSpans(), trace, func(span tracetest.SpanStub) {
if found.Name != "" {
if found != nil && found.Name != "" {
panic("found more than one span with the same trace string")
}
found = span
found = &span
})
return found
}
Expand Down Expand Up @@ -121,7 +121,7 @@ func (c Collector) SingleExceptionEvent(t *testing.T, trace string, typeRe strin
// has ContextCancelError exception recorded in the right place
et := c.FindSpanByTraceString(trace)
require.Len(t, et.Events, 1, "expected one event in span %v", trace)
ex := EventAsException(t, EventInTraceSpan(t, et, "exception"))
ex := EventAsException(t, EventInTraceSpan(t, *et, "exception"))
require.Regexp(t, typeRe, ex.Type)
require.Regexp(t, messageRe, ex.Message)
if errorCode {
Expand Down