Fix span error tagging in GO YARPC/gRPC

api/transport/propagation.go

@@ -27,6 +27,7 @@ import (
 	"github.com/opentracing/opentracing-go"
 	"github.com/opentracing/opentracing-go/ext"
 	opentracinglog "github.com/opentracing/opentracing-go/log"
+	"go.uber.org/yarpc/yarpcerrors"
 )
 
 // CreateOpenTracingSpan creates a new context with a started span
@@ -37,6 +38,11 @@ type CreateOpenTracingSpan struct {
 	ExtraTags     opentracing.Tags
 }
 
+const (
+	//TracingTagStatusCode is the span tag key for the YAPRC status code.
+	TracingTagStatusCode = "rpc.yarpc.status_code"
+)
+
 // Do creates a new context that has a reference to the started span.
 // This should be called before a Outbound makes a call
 func (c *CreateOpenTracingSpan) Do(
@@ -112,10 +118,11 @@ func (e *ExtractOpenTracingSpan) Do(
 
 // UpdateSpanWithErr sets the error tag on a span, if an error is given.
 // Returns the given error
-func UpdateSpanWithErr(span opentracing.Span, err error) error {
+func UpdateSpanWithErr(span opentracing.Span, err error, errCode yarpcerrors.Code) error {
 	if err != nil {
 		span.SetTag("error", true)
 		span.LogFields(opentracinglog.String("event", err.Error()))
+		span.SetTag(TracingTagStatusCode, errCode)
 	}
 	return err
 }

tracing.go

@@ -26,8 +26,14 @@ import (
 	opentracing "github.com/opentracing/opentracing-go"
 )
 
+const (
+	//TracingComponentName helps determine the attribution of a span.
+	TracingComponentName = "yarpc"
+)
+
 // OpentracingTags are tags with YARPC metadata.
 var OpentracingTags = opentracing.Tags{
 	"yarpc.version": Version,
 	"go.version":    runtime.Version(),
+	"component":     TracingComponentName,
 }

transport/grpc/handler.go

@@ -149,15 +149,15 @@ func (h *handler) handleStream(
 	stream := newServerStream(ctx, &transport.StreamRequest{Meta: transportRequest.ToRequestMeta()}, serverStream)
 	tServerStream, err := transport.NewServerStream(stream)
 	if err != nil {
-		return err
+		return transport.UpdateSpanWithErr(span, err, yarpcerrors.FromError(err).Code())
 	}
 	apperr := transport.InvokeStreamHandler(transport.StreamInvokeRequest{
 		Stream:  tServerStream,
 		Handler: streamHandler,
 		Logger:  h.logger,
 	})
 	apperr = handlerErrorToGRPCError(apperr, nil)
-	return transport.UpdateSpanWithErr(span, apperr)
+	return transport.UpdateSpanWithErr(span, apperr, yarpcerrors.FromError(err).Code())
 }
 
 func (h *handler) handleUnary(
@@ -230,7 +230,7 @@ func (h *handler) handleUnaryBeforeErrorConversion(
 	defer span.Finish()
 
 	err := h.callUnary(ctx, transportRequest, handler, responseWriter)
-	return transport.UpdateSpanWithErr(span, err)
+	return transport.UpdateSpanWithErr(span, err, yarpcerrors.FromError(err).Code())
 }
 
 func (h *handler) callUnary(ctx context.Context, transportRequest *transport.Request, unaryHandler transport.UnaryHandler, responseWriter *responseWriter) error {

transport/grpc/outbound.go

@@ -161,18 +161,27 @@ func (o *Outbound) invoke(
 	responseMD *metadata.MD,
 	start time.Time,
 ) (retErr error) {
+	tracer := o.t.options.tracer
+	createOpenTracingSpan := &transport.CreateOpenTracingSpan{
+		Tracer:        tracer,
+		TransportName: TransportName,
+		StartTime:     start,
+		ExtraTags:     yarpc.OpentracingTags,
+	}
+	ctx, span := createOpenTracingSpan.Do(ctx, request)
+	defer span.Finish()
 	md, err := transportRequestToMetadata(request)
 	if err != nil {
-		return err
+		return transport.UpdateSpanWithErr(span, err, yarpcerrors.FromError(err).Code())
 	}
 
 	bytes, err := ioutil.ReadAll(request.Body)
 	if err != nil {
-		return err
+		return transport.UpdateSpanWithErr(span, err, yarpcerrors.FromError(err).Code())
 	}
 	fullMethod, err := procedureNameToFullMethod(request.Procedure)
 	if err != nil {
-		return err
+		return transport.UpdateSpanWithErr(span, err, yarpcerrors.FromError(err).Code())
 	}
 	var callOptions []grpc.CallOption
 	if responseMD != nil {
@@ -183,7 +192,7 @@ func (o *Outbound) invoke(
 	}
 	apiPeer, onFinish, err := o.peerChooser.Choose(ctx, request)
 	if err != nil {
-		return err
+		return transport.UpdateSpanWithErr(span, err, yarpcerrors.FromError(err).Code())
 	}
 	defer func() { onFinish(retErr) }()
 	grpcPeer, ok := apiPeer.(*grpcPeer)
@@ -194,38 +203,26 @@ func (o *Outbound) invoke(
 		}
 	}
 
-	tracer := o.t.options.tracer
-	createOpenTracingSpan := &transport.CreateOpenTracingSpan{
-		Tracer:        tracer,
-		TransportName: TransportName,
-		StartTime:     start,
-		ExtraTags:     yarpc.OpentracingTags,
-	}
-	ctx, span := createOpenTracingSpan.Do(ctx, request)
-	defer span.Finish()
-
 	if err := tracer.Inject(span.Context(), opentracing.HTTPHeaders, mdReadWriter(md)); err != nil {
 		return err
 	}
 
-	err = transport.UpdateSpanWithErr(
-		span,
-		grpcPeer.clientConn.Invoke(
-			metadata.NewOutgoingContext(ctx, md),
-			fullMethod,
-			bytes,
-			responseBody,
-			callOptions...,
-		),
-	)
-	if err != nil {
+	if err := grpcPeer.clientConn.Invoke(
+		metadata.NewOutgoingContext(ctx, md),
+		fullMethod,
+		bytes,
+		responseBody,
+		callOptions...,
+	); err != nil {
+		err := transport.UpdateSpanWithErr(span, err, yarpcerrors.FromError(err).Code())
 		return invokeErrorToYARPCError(err, *responseMD)
 	}
+
 	// Service name match validation, return yarpcerrors.CodeInternal error if not match
 	if match, resSvcName := checkServiceMatch(request.Service, *responseMD); !match {
 		// If service doesn't match => we got response => span must not be nil
 		return transport.UpdateSpanWithErr(span, yarpcerrors.InternalErrorf("service name sent from the request "+
-			"does not match the service name received in the response: sent %q, got: %q", request.Service, resSvcName))
+			"does not match the service name received in the response: sent %q, got: %q", request.Service, resSvcName), yarpcerrors.CodeInternal)
 	}
 	return nil
 }
@@ -297,23 +294,32 @@ func (o *Outbound) stream(
 		return nil, yarpcerrors.InvalidArgumentErrorf("stream request requires a request metadata")
 	}
 	treq := req.Meta.ToRequest()
+	tracer := o.t.options.tracer
+	createOpenTracingSpan := &transport.CreateOpenTracingSpan{
+		Tracer:        tracer,
+		TransportName: TransportName,
+		StartTime:     start,
+		ExtraTags:     yarpc.OpentracingTags,
+	}
+	_, span := createOpenTracingSpan.Do(ctx, treq)
+	defer span.Finish()
 	if err := validateRequest(treq); err != nil {
-		return nil, err
+		return nil, transport.UpdateSpanWithErr(span, err, yarpcerrors.FromError(err).Code())
 	}
 
 	md, err := transportRequestToMetadata(treq)
 	if err != nil {
-		return nil, err
+		return nil, transport.UpdateSpanWithErr(span, err, yarpcerrors.FromError(err).Code())
 	}
 
 	fullMethod, err := procedureNameToFullMethod(req.Meta.Procedure)
 	if err != nil {
-		return nil, err
+		return nil, transport.UpdateSpanWithErr(span, err, yarpcerrors.FromError(err).Code())
 	}
 
 	apiPeer, onFinish, err := o.peerChooser.Choose(ctx, treq)
 	if err != nil {
-		return nil, err
+		return nil, transport.UpdateSpanWithErr(span, err, yarpcerrors.FromError(err).Code())
 	}
 
 	grpcPeer, ok := apiPeer.(*grpcPeer)
@@ -323,22 +329,13 @@ func (o *Outbound) stream(
 			ExpectedType: "*grpcPeer",
 		}
 		onFinish(err)
-		return nil, err
-	}
-
-	tracer := o.t.options.tracer
-	createOpenTracingSpan := &transport.CreateOpenTracingSpan{
-		Tracer:        tracer,
-		TransportName: TransportName,
-		StartTime:     start,
-		ExtraTags:     yarpc.OpentracingTags,
+		return nil, transport.UpdateSpanWithErr(span, err, yarpcerrors.FromError(err).Code())
 	}
-	_, span := createOpenTracingSpan.Do(ctx, treq)
 
 	if err := tracer.Inject(span.Context(), opentracing.HTTPHeaders, mdReadWriter(md)); err != nil {
 		span.Finish()
 		onFinish(err)
-		return nil, err
+		return nil, transport.UpdateSpanWithErr(span, err, yarpcerrors.FromError(err).Code())
 	}
 
 	streamCtx := metadata.NewOutgoingContext(ctx, md)

transport/grpc/stream.go

@@ -174,7 +174,7 @@ func (cs *clientStream) Headers() (transport.Headers, error) {
 
 func (cs *clientStream) closeWithErr(err error) error {
 	if !cs.closed.Swap(true) {
-		err = transport.UpdateSpanWithErr(cs.span, err)
+		err = transport.UpdateSpanWithErr(cs.span, err, yarpcerrors.FromError(err).Code())
 		cs.span.Finish()
 		cs.release(err)
 	}

transport/http/outbound.go

@@ -328,7 +328,7 @@ func (o *Outbound) call(ctx context.Context, treq *transport.Request) (*transpor
 		}
 		return nil, transport.UpdateSpanWithErr(span,
 			yarpcerrors.InternalErrorf("service name sent from the request "+
-				"does not match the service name received in the response, sent %q, got: %q", treq.Service, resSvcName))
+				"does not match the service name received in the response, sent %q, got: %q", treq.Service, resSvcName), yarpcerrors.CodeInternal)
 	}
 
 	tres := &transport.Response{