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(
@@ -110,12 +116,25 @@ func (e *ExtractOpenTracingSpan) Do(
  return ctx, span
 }
 
-// UpdateSpanWithErr sets the error tag on a span, if an error is given.
-// Returns the given error
+// UpdateSpanWithErr logs an error to the span. Prefer UpdateSpanWithErrAndCode
+// for including an error code in addition to the error message.
+// Deprecated: Use UpdateSpanWithErrAndCode instead.
 func UpdateSpanWithErr(span opentracing.Span, err error) error {
  if err != nil {
  span.SetTag("error", true)
- span.LogFields(opentracinglog.String("event", err.Error()))
+ span.LogFields(
+ opentracinglog.String("event", "error"),
+ opentracinglog.String("message", err.Error()),
+ )
  }
  return err
 }
+
+// UpdateSpanWithErrAndCode sets the error tag with errcode on a span, if an error is given.
+// Returns the given error
+func UpdateSpanWithErrAndCode(span opentracing.Span, err error, errCode yarpcerrors.Code) error {
+ if err != nil {
+ span.SetTag(TracingTagStatusCode, errCode)
+ }
+ return UpdateSpanWithErr(span, 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.UpdateSpanWithErrAndCode(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.UpdateSpanWithErrAndCode(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.UpdateSpanWithErrAndCode(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.UpdateSpanWithErrAndCode(span, err, yarpcerrors.FromError(err).Code())
  }
 
  bytes, err := ioutil.ReadAll(request.Body)
  if err != nil {
- return err
+ return transport.UpdateSpanWithErrAndCode(span, err, yarpcerrors.FromError(err).Code())
  }
  fullMethod, err := procedureNameToFullMethod(request.Procedure)
  if err != nil {
- return err
+ return transport.UpdateSpanWithErrAndCode(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.UpdateSpanWithErrAndCode(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.UpdateSpanWithErrAndCode(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))
+ return transport.UpdateSpanWithErrAndCode(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), 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.UpdateSpanWithErrAndCode(span, err, yarpcerrors.FromError(err).Code())
  }
 
  md, err := transportRequestToMetadata(treq)
  if err != nil {
- return nil, err
+ return nil, transport.UpdateSpanWithErrAndCode(span, err, yarpcerrors.FromError(err).Code())
  }
 
  fullMethod, err := procedureNameToFullMethod(req.Meta.Procedure)
  if err != nil {
- return nil, err
+ return nil, transport.UpdateSpanWithErrAndCode(span, err, yarpcerrors.FromError(err).Code())
  }
 
  apiPeer, onFinish, err := o.peerChooser.Choose(ctx, treq)
  if err != nil {
- return nil, err
+ return nil, transport.UpdateSpanWithErrAndCode(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.UpdateSpanWithErrAndCode(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.UpdateSpanWithErrAndCode(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.UpdateSpanWithErrAndCode(cs.span, err, yarpcerrors.FromError(err).Code())
  cs.span.Finish()
  cs.release(err)
  }

transport/http/outbound.go

@@ -326,9 +326,9 @@ func (o *Outbound) call(ctx context.Context, treq *transport.Request) (*transpor
  if err = response.Body.Close(); err != nil {
  return nil, yarpcerrors.Newf(yarpcerrors.CodeInternal, err.Error())
  }
- return nil, transport.UpdateSpanWithErr(span,
+ return nil, transport.UpdateSpanWithErrAndCode(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{