Go OpenTelemetry - 同一服务中 gRPC 服务器和客户端之间的相关跨度如何

问题描述

我有一个公开 gRPC 服务的 go 服务，并使用另一个 gRPC 服务，最后一个是基于 gRPC java 的服务，它使用 opentelemetry 正确跟踪。

在我的 go 服务中，我从 repo opentelemetry-go-contrib 复制了这个文件：interceptor.go 和 grpctrace.go，这里 https://github.com/open-telemetry/opentelemetry-go-contrib/tree/main/instrumentation/google.golang.org/grpc/otelgrpc

现在，我创建一个名为 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{}))
}

现在，当我启动我的 gRPC 服务器时，我会这样做：

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

这是服务器拦截器，每个请求都会调用它：

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,s.Proto())
    } else {
      span.SetAttributes(statusCodeAttr(grpc_codes.OK))
      messageSent.Event(ctx,resp)
    }

    return resp,err
  }
}

消费gRPC java服务的客户端：

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
}

还有，客户端拦截器：

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,attr := spanInfo(method,cc.Target())
    var span trace.Span
    ctx,span = tracer.Start(
      ctx,trace.WithSpanKind(trace.SpanKindClient),opts...)
    ctx = Metadata.NewOutgoingContext(ctx,Metadatacopy)

    messageSent.Event(ctx,req)

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

    messageReceived.Event(ctx,reply)

    if err != nil {
      s,s.Message())
      span.SetAttributes(statusCodeAttr(s.Code()))
    } else {
      span.SetAttributes(statusCodeAttr(grpc_codes.OK))
    }

    return err
  }
}

这会生成两个 span，第一个与服务器请求和响应，第二个与客户端信息，但这不相关。

我应该怎么做才能使两个拦截器的跨度相关？

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）