How correlated span between gRPC server and client in same service

[victorpacheco3107]

Hi!

I have a go service that expose a gRPC service, and consume another gRPC service, this last is a gRPC java based service, that tracing correctly with opentelemetry.

In my go service, I copy this files: interceptor.go and grpctrace.go from repo opentelemetry-go-contrib, here https://github.com/open-telemetry/opentelemetry-go-contrib/tree/main/instrumentation/google.golang.org/grpc/otelgrpc

Now, I create a file called config.go:

package grpcTracing

import (
  "context"
  "log"

  "go.opentelemetry.io/otel"
  "go.opentelemetry.io/otel/exporters/otlp"
  "go.opentelemetry.io/otel/exporters/otlp/otlpgrpc"
  "go.opentelemetry.io/otel/label"
  "go.opentelemetry.io/otel/propagation"
  "go.opentelemetry.io/otel/sdk/resource"
  sdktrace "go.opentelemetry.io/otel/sdk/trace"
)

// Init configures an OpenTelemetry exporter and trace provider
func Init() {

  ctx := context.Background()

  driver := otlpgrpc.NewDriver(
    otlpgrpc.WithInsecure(),
    otlpgrpc.WithEndpoint("localhost:55680"),
  )

  exporter, err := otlp.NewExporter(ctx, driver) // Configure as needed.
  if err != nil {
    log.Fatal(err)
  }

  service := "test-service"

  tracerProvider := sdktrace.NewTracerProvider(
    sdktrace.WithConfig(sdktrace.Config{DefaultSampler: sdktrace.AlwaysSample()}),
    sdktrace.WithResource(resource.NewWithAttributes(
      label.Key("service.name").String(service),
    )),
    sdktrace.WithBatcher(exporter),
  )

  otel.SetTracerProvider(tracerProvider)
  otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{}))
}

Now, when I start my gRPC server, I do this:

        grpcTracing.InitTracing()
        ...
	grpcServer := grpc.NewServer(
		grpc.UnaryInterceptor(grpcTracing.UnaryServerInterceptor()),
	)

This is the server interceptor, this is called every petition:

func UnaryServerInterceptor(opts ...Option) grpc.UnaryServerInterceptor {
  return func(
    ctx context.Context,
    req interface{},
    info *grpc.UnaryServerInfo,
    handler grpc.UnaryHandler,
  ) (interface{}, error) {
    requestMetadata, _ := metadata.FromIncomingContext(ctx)
    metadataCopy := requestMetadata.Copy()

    entries, spanCtx := Extract(ctx, &metadataCopy, opts...)
    ctx = baggage.ContextWithValues(ctx, entries...)

    tracer := newConfig(opts).TracerProvider.Tracer(
      instrumentationName,
      trace.WithInstrumentationVersion(otelcontrib.SemVersion()),
    )

    name, attr := spanInfo(info.FullMethod, peerFromCtx(ctx))

    ctx, span := tracer.Start(
      trace.ContextWithRemoteSpanContext(ctx, spanCtx),
      name,
      trace.WithSpanKind(trace.SpanKindServer),
      trace.WithAttributes(attr...),
    )
    defer span.End()

    Inject(ctx, &requestMetadata, opts...)
    ctx = metadata.NewIncomingContext(ctx, metadataCopy)

    messageReceived.Event(ctx, 1, req)

    span.SetAttributes(label.Any("request", req))

    resp, err := handler(ctx, req)

    span.SetAttributes(label.Any("response", resp))

    if err != nil {
      s, _ := status.FromError(err)
      span.SetStatus(codes.Error, s.Message())
      span.SetAttributes(statusCodeAttr(s.Code()))
      messageSent.Event(ctx, 1, s.Proto())
    } else {
      span.SetAttributes(statusCodeAttr(grpc_codes.OK))
      messageSent.Event(ctx, 1, resp)
    }

    return resp, err
  }
}

The client that consume the gRPC java service:

func (g *GrpcAndesClient) FindDigitalCertificate(transaction *entities.Transaction, customer *entities.Customer) ([]entities.Certificate, error) {
	var conn *grpc.ClientConn
	var err error
	if g.Client == nil {
		var opts []grpc.DialOption
		opts = append(opts, grpc.WithInsecure(), grpc.WithUnaryInterceptor(grpcTracing.UnaryClientInterceptor())) // <--Client interceptor
		conn, err = grpc.Dial(g.endPoint, opts...)
		if err != nil {
			log.Fatalf(dialGrpcServiceError, err)
		}
		g.Client = generatedClient.NewAndesClientAppClient(conn)
		defer conn.Close()
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(g.timeout)*time.Second)
	defer cancel()

	request := BuildFindCertificateClientRequest(transaction, customer)

	md := metadata.Pairs()

	ctx = metadata.NewOutgoingContext(context.Background(), md)

	clientResponse, err := g.Client.FindDigitalCertificate(ctx, request)
	if err != nil {
		g.Client = nil
		return make([]entities.Certificate, 0),
			errcatalogs.MakeBusinessResponseError(
				http.StatusPartialContent, fmt.Sprintf(andesClientFindCertificatesErrorMessage, err.Error()))
	}
	responseCertificates := BuildResponseCertificates(clientResponse)
	g.Client = nil
	return responseCertificates, nil
}

And, the client interceptor:

func UnaryClientInterceptor(opts ...Option) grpc.UnaryClientInterceptor {
  return func(
    ctx context.Context,
    method string,
    req, reply interface{},
    cc *grpc.ClientConn,
    invoker grpc.UnaryInvoker,
    callOpts ...grpc.CallOption,
  ) error {

    requestMetadata, _ := metadata.FromOutgoingContext(ctx)
    metadataCopy := requestMetadata.Copy()


    tracer := newConfig(opts).TracerProvider.Tracer(
      instrumentationName,
      trace.WithInstrumentationVersion(otelcontrib.SemVersion()),
    )

    name, attr := spanInfo(method, cc.Target())
    var span trace.Span
    ctx, span = tracer.Start(
      ctx,
      name,
      trace.WithSpanKind(trace.SpanKindClient),
      trace.WithAttributes(attr...),
    )
    defer span.End()

    Inject(ctx, &metadataCopy, opts...)
    ctx = metadata.NewOutgoingContext(ctx, metadataCopy)

    messageSent.Event(ctx, 1, req)

    err := invoker(ctx, method, req, reply, cc, callOpts...)

    messageReceived.Event(ctx, 1, reply)

    if err != nil {
      s, _ := status.FromError(err)
      span.SetStatus(codes.Error, s.Message())
      span.SetAttributes(statusCodeAttr(s.Code()))
    } else {
      span.SetAttributes(statusCodeAttr(grpc_codes.OK))
    }

    return err
  }
}

This generate two span, the first with server request and response, the second with the client info, but this are not correlated.

What should I do so that the spans of the two interceptors are correlated?