Response message tracing (#327)

* feat(queryexecutor): trace per block spans

* feat(responseassembler): add timings in response assembler

* feat(messagequeue): add message tracing

* test(fix tests and cleanup):

N

* style(lint): fix staticcheck

* test(impl): fix flaky test

impossible to know if one or two messages will be sent

* feat(messagequeue): record error

* fix(graphsync): test cleanups and refactors

* fix(impl): fix data race

Author: hannahhoward

messagequeue/builder.go

@@ -1,6 +1,7 @@
 package messagequeue
 
 import (
+	"context"
 	"io"
 
 	"github.com/ipfs/go-graphsync"
@@ -11,6 +12,7 @@ import (
 // Builder wraps a message builder with additional functions related to metadata
 // and notifications in the message queue
 type Builder struct {
+	ctx context.Context
 	*gsmsg.Builder
 	topic           Topic
 	responseStreams map[graphsync.RequestID]io.Closer
@@ -19,8 +21,9 @@ type Builder struct {
 }
 
 // NewBuilder sets up a new builder for the given topic
-func NewBuilder(topic Topic) *Builder {
+func NewBuilder(ctx context.Context, topic Topic) *Builder {
 	return &Builder{
+		ctx:             ctx,
 		Builder:         gsmsg.NewBuilder(),
 		topic:           topic,
 		responseStreams: make(map[graphsync.RequestID]io.Closer),
@@ -29,6 +32,10 @@ func NewBuilder(topic Topic) *Builder {
 	}
 }
 
+func (b *Builder) Context() context.Context {
+	return b.ctx
+}
+
 // SetResponseStream sets the given response stream to close should the message fail to send
 func (b *Builder) SetResponseStream(requestID graphsync.RequestID, stream io.Closer) {
 	b.responseStreams[requestID] = stream
@@ -82,6 +89,7 @@ func (b *Builder) build(publisher notifications.Publisher) (gsmsg.GraphSyncMessa
 			BlockData:     b.blockData,
 			ResponseCodes: message.ResponseCodes(),
 		},
+		ctx:             b.ctx,
 		topic:           b.topic,
 		msgSize:         b.BlockSize(),
 		responseStreams: b.responseStreams,

messagequeue/messagequeue.go

@@ -11,6 +11,10 @@ import (
 	"github.com/ipfs/go-graphsync"
 	logging "github.com/ipfs/go-log/v2"
 	"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"
 
 	gsmsg "github.com/ipfs/go-graphsync/message"
 	gsnet "github.com/ipfs/go-graphsync/network"
@@ -112,7 +116,10 @@ func (mq *MessageQueue) buildMessage(size uint64, buildMessageFn func(*Builder))
 	if shouldBeginNewResponse(mq.builders, size) {
 		topic := mq.nextBuilderTopic
 		mq.nextBuilderTopic++
-		mq.builders = append(mq.builders, NewBuilder(topic))
+		ctx, _ := otel.Tracer("graphsync").Start(mq.ctx, "message", trace.WithAttributes(
+			attribute.Int64("topic", int64(topic)),
+		))
+		mq.builders = append(mq.builders, NewBuilder(ctx, topic))
 	}
 	builder := mq.builders[len(mq.builders)-1]
 	buildMessageFn(builder)
@@ -156,7 +163,12 @@ func (mq *MessageQueue) runQueue() {
 				for {
 					_, metadata, err := mq.extractOutgoingMessage()
 					if err == nil {
-						mq.publishError(metadata, fmt.Errorf("message queue shutdown"))
+						span := trace.SpanFromContext(metadata.ctx)
+						err := fmt.Errorf("message queue shutdown")
+						span.RecordError(err)
+						span.SetStatus(codes.Error, err.Error())
+						span.End()
+						mq.publishError(metadata, err)
 						mq.eventPublisher.Close(metadata.topic)
 					} else {
 						break
@@ -211,12 +223,20 @@ func (mq *MessageQueue) extractOutgoingMessage() (gsmsg.GraphSyncMessage, intern
 
 func (mq *MessageQueue) sendMessage() {
 	message, metadata, err := mq.extractOutgoingMessage()
+
 	if err != nil {
 		if err != errEmptyMessage {
 			log.Errorf("Unable to assemble GraphSync message: %s", err.Error())
 		}
 		return
 	}
+	span := trace.SpanFromContext(metadata.ctx)
+	defer span.End()
+	_, sendSpan := otel.Tracer("graphsync").Start(metadata.ctx, "sendMessage", trace.WithAttributes(
+		attribute.Int64("topic", int64(metadata.topic)),
+		attribute.Int64("size", int64(metadata.msgSize)),
+	))
+	defer sendSpan.End()
 	mq.publishQueued(metadata)
 	defer mq.eventPublisher.Close(metadata.topic)
 
@@ -337,6 +357,7 @@ func openSender(ctx context.Context, network MessageNetwork, p peer.ID, sendTime
 }
 
 type internalMetadata struct {
+	ctx             context.Context
 	public          Metadata
 	topic           Topic
 	msgSize         uint64

messagequeue/messagequeue_test.go

@@ -183,6 +183,7 @@ func TestProcessingNotification(t *testing.T) {
 
 func TestDedupingMessages(t *testing.T) {
 	ctx := context.Background()
+	ctx, collectTracing := testutil.SetupTracing(ctx)
 	ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
 	defer cancel()
 
@@ -251,6 +252,12 @@ func TestDedupingMessages(t *testing.T) {
 			t.Fatal("incorrect request added to message")
 		}
 	}
+
+	tracing := collectTracing(t)
+	require.ElementsMatch(t, []string{
+		"message(0)->sendMessage(0)",
+		"message(1)->sendMessage(0)",
+	}, tracing.TracesToStrings())
 }
 
 func TestSendsVeryLargeBlocksResponses(t *testing.T) {