• OpenTelemetry 深度定制:跨服务追踪的实战技巧


    背景

    在上一篇《从 Dapper 到 OpenTelemetry:分布式追踪的演进之旅》中在最后提到在做一些 Trace 的定制开发。

    到现在差不多算是完成了,可以和大家分享一下。

    我们的需求是这样的:


    假设现在有三个服务:ServiceA、ServiceB、ServiceC

    ServiceA 对外提供了一个 http 接口 request,在这个接口会调用 ServiceBorder 订单接口创建订单,同时 serviceB 调用 serviceC 的 pay 接口。


    整个调用关系如上图所示。

    默认情况下 span 中的 attribute 会记录当前 span 的一些信息,比如:

    这些都是当前一些当前 span 内置的信息,比如当前 gRPC 接口的一些基本数据:服务名、ip、端口等信息。

    但这里并没有上游的一些信息,虽然我们可以通过 Jaeger 的树状图得知上游是哪个应用调用过来的,但是一旦某个 span 下有多个子 span 的调用,就没办法很直观知道这个子 span 的上游是由谁发起的调用。

    比如如下这个链路:

    当一个调用链非常长,同时也非常复杂时,没办法第一时间知道某一个 span 的上游到底是谁发起的,需要手动一层层的去折叠,或者全靠眼睛去找。

    预期效果

    为此我们希望的效果是可以通过给每一个子 span 中加入两个 attribute,来标明它的父调用来源。

    比如在 serviceB 中的所有 span 中都会加上两个标签:来源是 serviceA,同时是 serviceA 的 request 接口发起的请求。

    而在 serviceC 中同样可以知道来源是 serviceB 的 Order 接口发起的调用。

    我启动了三个 demo 应用,分别是 create1,create2,create3.

    create1 中会提供一个 request 接口,在这里面调用 create2 的 create2 接口,create2 的接口里接着调用 create3 的 create3 接口。

    create1:

        @RequestMapping("/request")  
        public String request(@RequestParam String name) {  
           HelloRequest request = HelloRequest.newBuilder()  
                 .setName(name)  
                 .build();  
           log.info("request: {}", request);  
           String message = myServiceStub.create2(request).getMessage();  
           Executors.newFixedThreadPool(1).execute(() -> {  
              myServiceStub.create2(request).getMessage();  
           });       return message;  
        }
    

    create2:

    @Override  
    public void create2(HelloRequest request, StreamObserver responseObserver) {  
        HelloReply reply = HelloReply.newBuilder()  
                .setMessage("Create2 ==> " + request.getName())  
                .build();  
        log.info("Create2: {}", reply.getMessage());  
        myMethod(request.getName());  
        myServiceStub.create3(request);
        responseObserver.onNext(reply);  
        responseObserver.onCompleted();  
    }
    

    create3:

    @Override  
    public void create3(HelloRequest request, StreamObserver responseObserver) {  
        HelloReply reply = HelloReply.newBuilder()  
                .setMessage("Create3 ==> " + request.getName())  
                .build();  
        log.info("Create3: {}", reply.getMessage());  
        myMethod(request.getName());  
        responseObserver.onNext(reply);  
        responseObserver.onCompleted();  
    }
    
    java -javaagent:opentelemetry-javaagent-2.4.0-SNAPSHOT.jar \
    -Dotel.javaagent.extensions=otel-extensions-custom-context-1.0-SNAPSHOT.jar \
    -Dotel.traces.exporter=otlp \
    -Dotel.logs.exporter=none \
    -Dotel.service.name=create2 \
    -Dotel.exporter.otlp.protocol=grpc \
    -Dotel.propagators=tracecontext,baggage,demo \
    -Dotel.exporter.otlp.endpoint=http://127.0.0.1:5317 \
          -jar target/demo-0.0.1-SNAPSHOT.jar --spring.application.name=create2 --server.port=9191 --grpc.server.port=9292 --grpc.client.myService.address=static://127.0.0.1:9393
    

    只是每个应用都需要使用我这边单独打的 agent 包以及一个 extension(tel-extensions-custom-context-1.0-SNAPSHOT.jar) 才能生效。

    最终的效果如下:

    Baggage

    在讲具体的实现之前需要先了解几个 Trace 中的概念,在这里主要用到的是一个称为 Baggage 的对象。

    在之前的文章中其实提到过它的原理以及使用场景:
    从 Dapper 到 OpenTelemetry:分布式追踪的演进之旅

    Baggage 的中文翻译是:包裹📦;简单来说就是我们可以通过自定义 baggage 可以将我们想要的数据存放在其中,这样再整个 Trace 的任意一个 Span 中都可以读取到。

    @RequestMapping("/request")  
    public String request(@RequestParam String name) {  
    	// 写入
        Baggage.current().toBuilder().  
              put("request.name", name).build()  
              .storeInContext(Context.current()).makeCurrent();
    }         
    
    // 获取
    String value = Baggage.current().getEntryValue("request.name");  
    log.info("request.name: {}", value);
    

    理解了这个之后,我们要实现的将上游的信息传递到下游就可以通过这个组件实现了。

    只需要在上游创建 span 时将它自身数据写入到 Baggage 中,再到下游 span 取出来写入到 attribute 中即可。

    ContextCustomizer

    这里的关键就是在哪里写入这个 Baggage,因为对第三方组件的 Instrumentation 的实现都是在 opentelemetry-java-instrumentation项目中。

    javaagent.jar 包也是通过该项目打包出来的。

    所以在该项目的 io.opentelemetry.instrumentation.api.instrumenter.Instrumenter#doStart 这个函数中我们发现一段逻辑:


    这个函数是在创建一个 span 的时候调用的,通常这个创建函数是在这些第三方库的拦截器中创建的。


    比如这是在 grpc 的拦截器中调用。

    // context customizers run before span start, so that they can have access to the parent span  
    // context, and so that their additions to the context will be visible to span processors  
    for (ContextCustomizersuper REQUEST> contextCustomizer : contextCustomizers) {  
      context = contextCustomizer.onStart(context, request, attributes);  
    }
    

    ContextCustomizer 是一个接口只提供了一个函数:

    public interface ContextCustomizer {  
      
      /** Allows to customize the operation {@link Context}. */  
      Context onStart(Context parentContext, REQUEST request, Attributes startAttributes);  
    }
    
    • Context 是上下文信息,可以在自定义的 ContextCustomizer 继续往上下文中追加信息。
    • REQUEST 是一个泛型:一般是当前第三方组件的请求信息:
      • 比如是 HTTP 时,这个 request 就是 HTTP 的请求信息。
      • 而如果是 gRPC ,则是 gRPC 的请求信息。
      • 其他的请求类型同理。
    • startAttributes 则是预先写入的一些属性,比如在上图中看到的一些 rpc.service/rpc.method等字段。
    // context customizers run before span start, so that they can have access to the parent span  
    // context, and so that their additions to the context will be visible to span processors
    

    从这个接口的调用注释可以看出:
    这个自定义的 context 会在 span 开始之前调用,所以在这里是可以访问到当前创建的 span 的父 context,同时在这里的 context 中新增的数据可以在 SpanProcessor 访问到。

    SpanProcessor

    而 SpanProcessor 又是一个非常的重要的组件,我们接着刚才的 contextCustomizer 处往后跟踪代码。

    context = contextCustomizer.onStart(context, request, attributes);
    	--->Span span = spanBuilder.setParent(context).startSpan();
    			--->io.opentelemetry.sdk.trace.SdkSpanBuilder#startSpan
    				--->io.opentelemetry.sdk.trace.SdkSpan#startSpan
    					--->spanProcessor.onStart(parentContext, span);
    

    可以看到 spanProcessor.onStart 这个函数会在 contextCustomizer 之后调用。

    
    /**  
     * SpanProcessor is the interface {@link SdkTracer} uses to allow synchronous hooks for when a  
     * {@code Span} is started or when a {@code Span} is ended.  
     */
     
    //==========================================================
    
    /**  
     * Called when a {@link io.opentelemetry.api.trace.Span} is started, if the {@link  
     * Span#isRecording()} returns true.  
     * * 

    This method is called synchronously on the execution thread, should not throw or block the * execution thread. * * @param parentContext the parent {@code Context} of the span that just started. * @param span the {@code Span} that just started. */void onStart(Context parentContext, ReadWriteSpan span);

    从注释中可以知道 SpanProcessor 是作为一个 span 的生命周期中的关键节点的 hook 函数。

    在这些函数中我们可以自定义一些 span 的数据,比如在 onStart 还可以往 span 中写入一些自定义的 attribute。

    这也是我们这次会用到的一个接口,我们的方案是:

    在 gRPC 构建 Instrument 时自定义一个 GrpcServerContextCustomizer ,在这个自定义的 ContextCustomizer 中写入一个 Baggage

    然后在 io.opentelemetry.sdk.trace.SpanProcessor#onStart 接口中取出这个 Baggage 写入到当前 span 的 attribute 中。

    这样我们就可以看到之前提到的那些数据上游信息了。

    为 gRPC 添加上下文

    先来看看如何为 gRPC 添加 Baggage

    我们先自定义一个 GrpcServerContextCustomizer 实现类:

    public class GrpcServerContextCustomizer implements ContextCustomizer {  
      private final String currentServiceName;  
      
      private static final String PARENT_RPC_KEY = "parent_rpc";  
      private static final String CURRENT_RPC_KEY = "current_rpc";  
      
      private static final String CURRENT_HTTP_URL_PATH = "current_http_url_path";  
      
      public GrpcServerContextCustomizer(String serviceName) {  
        this.currentServiceName = serviceName;  
      }  
      @Override  
      public Context onStart(Context parentContext, GrpcRequest grpcRequest,  
          Attributes startAttributeds) {  
        BaggageBuilder builder = Baggage.fromContext(parentContext).toBuilder();  
      
        String currentRpc = Baggage.fromContext(parentContext).getEntryValue(CURRENT_RPC_KEY);  
        String fullMethodName = startAttributeds.get(AttributeKey.stringKey("rpc.method"));  
        String rpcService = startAttributeds.get(AttributeKey.stringKey("rpc.service"));  
        // call from grpc  
        String method = rpcService + ":" + fullMethodName;  
        String baggageInfo = getBaggageInfo(currentServiceName, method);  
      
        String httpUrlPath = Baggage.fromContext(parentContext).getEntryValue(CURRENT_HTTP_URL_PATH);  
        if (!StringUtils.isNullOrEmpty(httpUrlPath)) {  
          // call from http  
          // currentRpc = currentRpc;  currentRpc = create1|GET:/request      // clear current_http_url_path      builder.put(CURRENT_HTTP_URL_PATH, "");  
        }  
        Baggage baggage = builder  
            .put(PARENT_RPC_KEY, currentRpc)  
            .put(CURRENT_RPC_KEY, baggageInfo)  
            .build();  
        return parentContext.with(baggage);  
      
      }  
      private static String getBaggageInfo(String serviceName, String method) {  
        if (StringUtils.isNullOrEmpty(serviceName)) {  
          return "";  
        }    return serviceName + "|" + method;  
      }  
    }
    

    从这个代码中可以看出,我们需要先从上下文中获取 CURRENT_RPC_KEY ,从而得知当前的 span 是不是 root span。

    所以我们其实是把当前的 span 信息作为一个 PARENT_RPC_KEY 写入到 Baggage 中。

    这样在 SpanProcessor 中便可以直接取出 PARENT_RPC_KEY 作为上游的信息写入 span 的 attribute 中。

        @Override  
        public void onStart(Context parentContext, ReadWriteSpan span) {
            String parentRpc = Baggage.fromContext(parentContext).getEntryValue("parent_rpc");  
            if (!StringUtils.isNullOrEmpty(parentRpc)) {  
                String[] split = parentRpc.split("\\|");  
                span.setAttribute("parent_rpc", parentRpc);  
                span.setAttribute("parent_service_name", split[0]);  
                span.setAttribute("parent_service_method", split[1]); 
            }  
        }
    

    需要注意的是,这里的 Baggage 需要使用 Baggage.fromContext(parentContext) 才能拿到刚才写入 Baggage 信息。

    之后我们找到构建 gRPCServerInstrumenterBuilder 的地方,写入我们刚才自定义的 GrpcServerContextCustomizer 即可。

    .addContextCustomizer(new GrpcServerContextCustomizer(serviceName))
    

    这里我们选择写入到是 serverInstrumenterBuilder 而不是clientInstrumenterBuilder,因为在服务端的入口就知道是从哪个接口进来的请求。

    为 spring boot 的 http 接口添加上下文

    如果只存在 gRPC 调用时只添加 gRPC 的上下文也够用了,但是我们也不排除由外部接口是通过 HTTP 访问进来的,然后再调用内部的 gRPC 接口;这也是非常常见的架构模式。

    所以我们还需要在 HTTP 中增加 ContextCustomizer 将自身的数据写入到 Baggage 中。

    好在 HttpServerRouteBuilder 自身是实现了 ContextCustomizer 接口的,我们只需要往里面写入 Baggage 数据即可。

    public ContextCustomizer build() {  
      Set knownMethods = new HashSet<>(this.knownMethods);  
      return (context, request, startAttributes) -> {  
        if (HttpRouteState.fromContextOrNull(context) != null) {  
          return context;  
        }    String method = getter.getHttpRequestMethod(request);  
        if (method == null || !knownMethods.contains(method)) {  
          method = "HTTP";  
        }    String urlPath = getter.getUrlPath(request);  
        String methodPath = method + ":" + urlPath;  
      
        String currentRpc = Baggage.fromContext(context).getEntryValue(CURRENT_RPC_KEY);  
        String baggageInfo = getBaggageInfo(serviceName, methodPath);  
        Baggage baggage = Baggage.fromContext(context).toBuilder()  
            .put(PARENT_RPC_KEY, currentRpc)  
            .put(CURRENT_RPC_KEY, baggageInfo)  
            .put(CURRENT_HTTP_URL_PATH, methodPath)  
            .build();   
        return context.with(HttpRouteState.create(method, null, 0))  
            .with(baggage);  
      };}
    

    这里新增了 CURRENT_HTTP_URL_PATH 用于标记当前的请求来源是 HTTP,在 grpc 的 ContextCustomizer 解析时会判断这个值是否为空。

    String httpUrlPath = Baggage.fromContext(parentContext).getEntryValue(CURRENT_HTTP_URL_PATH);  
    if (!StringUtils.isNullOrEmpty(httpUrlPath)) {  
      // call from http  
      // currentRpc = currentRpc;  currentRpc = create1|GET:/request  // clear current_http_url_path  builder.put(CURRENT_HTTP_URL_PATH, "");  
    }
    

    这样就可以在 grpc 的下游接口拿到入口的 HTTP 接口数据了。


    当然也有可能是在 grpc 接口中调用 HTTP 的接口的场景,只是我们的业务中没有这种情况,所以就没有适配这类的场景。

    总结

    ContextCustomizer 接口没有提供对应的扩展,但是 SpanProcessor 是提供了扩展接口的。

    原本是想尽量别维护自己的 javaagent,但也好在 OpenTelemetry 是提供的接口,所以也并不会去修改原本的代码。

    所以我们还是需要创建一个 extensions 的项目在实现 SpanProcessor,这个在之前的 《实战:如何编写一个 OpenTelemetry Extensions》有详细讲到。

    所以最后的应用启动方式如下:

    java -javaagent:opentelemetry-javaagent-2.4.0-SNAPSHOT.jar \
    -Dotel.javaagent.extensions=otel-extensions-custom-context-1.0-SNAPSHOT.jar \
    

    需要使用我们手动打包的 javaagent 以及一个自定义扩展包。

    打包方式:

     ./gradlew assemble
    

    opentelemetry-java-instrumentation 项目比较大,所以打包过程可能比较久。

    因为这其实是一些定制需求,所以就没提交到上游,感兴趣的可以自行合并代码测试。

    最后可以这个分支中查看到修改的部分:
    https://github.com/crossoverJie/opentelemetry-java-instrumentation/compare/main...add-grpc-context

  • 相关阅读:
    服务器(Linux)查看Tomcat运行日志
    3天带你走向实战,阿里顶配版Spring全家桶面试进阶笔记有多强?
    授权专利破百,通付盾潜心精研迎接数字时代新征程
    Java IO包中Reader及Writer的简介说明
    linux java 环境变量配置
    HTML5+CSS3+移动web 前端开发入门笔记(一)
    DP/typec转VGA(国产化,工业级)
    1353016-70-2,DBCO-COOH,DBCO-Acid,二苯并环辛炔-羧酸科研专用
    性能压力测试主要目标及步骤
    css外边距塌陷(合并)
  • 原文地址:https://www.cnblogs.com/crossoverJie/p/18270379