首页 > 其他分享 >OpenTelemetry 实战：gRPC 监控的实现原理

OpenTelemetry 实战：gRPC 监控的实现原理

时间：2024-09-04 09:54:01浏览次数：8  
标签：实战 context gRPC OpenTelemetry grpc io public

前言

最近在给 opentelemetry-java-instrumentation 提交了一个 PR，是关于给 gRPC 新增四个 metrics：

  • rpc.client.request.size: 客户端请求包大小
  • rpc.client.response.size：客户端收到的响应包大小
  • rpc.server.request.size：服务端收到的请求包大小
  • rpc.server.response.size：服务端响应的请求包大小

这个 PR 的主要目的就是能够在指标监控中拿到 RPC 请求的包大小，而这里的关键就是如何才能拿到这些包的大小。

首先支持的是 gRPC（目前在云原生领域使用的最多），其余的 RPC 理论上也是可以支持的：

在实现的过程中我也比较好奇 OpenTelemetry 框架是如何给 gRPC 请求创建 span 调用链的，如下图所示：
image.png
image.png

这是一个 gRPC 远程调用，java-demo 是 gRPC 的客户端，k8s-combat 是 gRPC 的服务端

在开始之前我们可以根据 OpenTelemetry 的运行原理大概猜测下它的实现过程。

首先我们应用可以创建这些链路信息的前提是：使用了 OpenTelemetry 提供的 javaagent，这个 agent 的原理是在运行时使用了 byte-buddy 增强了我们应用的字节码，在这些字节码中代理业务逻辑，从而可以在不影响业务的前提下增强我们的代码（只要就是创建 span、metrics 等数据）

Spring 的一些代理逻辑也是这样实现的

gRPC 增强原理

而在工程实现上，我们最好是不能对业务代码进行增强，而是要找到这些框架提供的扩展接口。

拿 gRPC 来说，我们可以使用它所提供的 io.grpc.ClientInterceptor 和 io.grpc.ServerInterceptor 接口来增强代码。

打开 io.opentelemetry.instrumentation.grpc.v1_6.TracingClientInterceptor 类我们可以看到它就是实现了 io.grpc.ClientInterceptor：

而其中最关键的就是要实现 io.grpc.ClientInterceptor#interceptCall 函数：

@Override  
public <REQUEST, RESPONSE> ClientCall<REQUEST, RESPONSE> interceptCall(  
    MethodDescriptor<REQUEST, RESPONSE> method, CallOptions callOptions, Channel next) {  
  GrpcRequest request = new GrpcRequest(method, null, null, next.authority());  
  Context parentContext = Context.current();  
  if (!instrumenter.shouldStart(parentContext, request)) {  
    return next.newCall(method, callOptions);  
  }  
  Context context = instrumenter.start(parentContext, request);  
  ClientCall<REQUEST, RESPONSE> result;  
  try (Scope ignored = context.makeCurrent()) {  
    try {  
      // call other interceptors  
      result = next.newCall(method, callOptions);  
    } catch (Throwable e) {  
      instrumenter.end(context, request, Status.UNKNOWN, e);  
      throw e;  
    }  }  
  return new TracingClientCall<>(result, parentContext, context, request);  
}

这个接口是 gRPC 提供的拦截器接口，对于 gRPC 客户端来说就是在发起真正的网络调用前后会执行的方法。

所以在这个接口中我们就可以实现创建 span 获取包大小等逻辑。

使用 byte-buddy 增强代码

不过有一个问题是我们实现的 io.grpc.ClientInterceptor 类需要加入到拦截器中才可以使用：

var managedChannel = ManagedChannelBuilder.forAddress(host, port) .intercept(new TracingClientInterceptor()) // 加入拦截器
.usePlaintext()
.build();

但在 javaagent 中是没法给业务代码中加上这样的代码的。

此时就需要 byte-buddy 登场了，它可以动态修改字节码从而实现类似于修改源码的效果。

在 io.opentelemetry.javaagent.instrumentation.grpc.v1_6.GrpcClientBuilderBuildInstr umentation 类里可以看到 OpenTelemetry 是如何使用 byte-buddy 的。

  @Override
  public ElementMatcher<TypeDescription> typeMatcher() {
    return extendsClass(named("io.grpc.ManagedChannelBuilder"))
        .and(declaresField(named("interceptors")));
  }

  @Override
  public void transform(TypeTransformer transformer) {
    transformer.applyAdviceToMethod(
        isMethod().and(named("build")),
        GrpcClientBuilderBuildInstrumentation.class.getName() + "$AddInterceptorAdvice");
  }

  @SuppressWarnings("unused")
  public static class AddInterceptorAdvice {

    @Advice.OnMethodEnter(suppress = Throwable.class)
    public static void addInterceptor(
        @Advice.This ManagedChannelBuilder<?> builder,
        @Advice.FieldValue("interceptors") List<ClientInterceptor> interceptors) {
      VirtualField<ManagedChannelBuilder<?>, Boolean> instrumented =
          VirtualField.find(ManagedChannelBuilder.class, Boolean.class);
      if (!Boolean.TRUE.equals(instrumented.get(builder))) {
        interceptors.add(0, GrpcSingletons.CLIENT_INTERCEPTOR);
        instrumented.set(builder, true);
      }
    }
  }

从这里的源码可以看出，使用了 byte-buddy 拦截了 io.grpc.ManagedChannelBuilder#intercept(java.util.List<io.grpc.ClientInterceptor>) 函数。

io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers#extendsClass/ isMethod 等函数都是 byte-buddy 库提供的函数。

而这个函数正好就是我们需要在业务代码里加入拦截器的地方。

interceptors.add(0, GrpcSingletons.CLIENT_INTERCEPTOR);
GrpcSingletons.CLIENT_INTERCEPTOR = new TracingClientInterceptor(clientInstrumenter, propagators);

通过这行代码可以手动将 OpenTelemetry 里的 TracingClientInterceptor 加入到拦截器列表中，并且作为第一个拦截器。

而这里的：

extendsClass(named("io.grpc.ManagedChannelBuilder"))
        .and(declaresField(named("interceptors")))

通过函数的名称也可以看出是为了找到 继承了io.grpc.ManagedChannelBuilder 类中存在成员变量 interceptors 的类。

transformer.applyAdviceToMethod(  
    isMethod().and(named("build")),  
    GrpcClientBuilderBuildInstrumentation.class.getName() + "$AddInterceptorAdvice");

然后在调用 build 函数后就会进入自定义的 AddInterceptorAdvice 类，从而就可以拦截到添加拦截器的逻辑，然后把自定义的拦截器加入其中。

获取 span 的 attribute

我们在 gRPC 的链路中还可以看到这个请求的具体属性，比如：

  • gRPC 服务提供的 IP 端口。
  • 请求的响应码
  • 请求的 service 和 method
  • 线程等信息。

这些信息在问题排查过程中都是至关重要的。

可以看到这里新的 attribute 主要是分为了三类：

  • net.* 是网络相关的属性
  • rpc.* 是和 grpc 相关的属性
  • thread.* 是线程相关的属性

所以理论上我们在设计 API 时最好可以将这些不同分组的属性解耦开，如果是 MQ 相关的可能还有一些 topic 等数据，所以各个属性之间是互不影响的。

带着这个思路我们来看看 gRPC 这里是如何实现的。

clientInstrumenterBuilder
	.setSpanStatusExtractor(GrpcSpanStatusExtractor.CLIENT)
	.addAttributesExtractors(additionalExtractors)
        .addAttributesExtractor(RpcClientAttributesExtractor.create(rpcAttributesGetter))
        .addAttributesExtractor(ServerAttributesExtractor.create(netClientAttributesGetter))
        .addAttributesExtractor(NetworkAttributesExtractor.create(netClientAttributesGetter))

OpenTelemetry 会提供一个 io.opentelemetry.instrumentation.api.instrumenter.InstrumenterBuilder#addAttributesExtractor构建器函数，用于存放自定义的属性解析器。

从这里的源码可以看出分别传入了网络相关、RPC 相关的解析器；正好也就对应了图中的那些属性，也满足了我们刚才提到的解耦特性。

而每一个自定义属性解析器都需要实现接口 io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor

public interface AttributesExtractor<REQUEST, RESPONSE> {
}

这里我们以 GrpcRpcAttributesGetter 为例。

enum GrpcRpcAttributesGetter implements RpcAttributesGetter<GrpcRequest> {
  INSTANCE;

  @Override
  public String getSystem(GrpcRequest request) {
    return "grpc";
  }

  @Override
  @Nullable
  public String getService(GrpcRequest request) {
    String fullMethodName = request.getMethod().getFullMethodName();
    int slashIndex = fullMethodName.lastIndexOf('/');
    if (slashIndex == -1) {
      return null;
    }
    return fullMethodName.substring(0, slashIndex);
  }

可以看到 system 是写死的 grpc，也就是对于到页面上的 rpc.system 属性。

而这里的 getService 函数则是拿来获取 rpc.service 属性的，可以看到它是通过 gRPC 的method 信息来获取 service 的。

public interface RpcAttributesGetter<REQUEST> {  
  
  @Nullable  
  String getService(REQUEST request);
}

而这里 REQUEST 其实是一个泛型，在 gRPC 里是 GrpcRequest，在其他 RPC 里这是对应的 RPC 的数据。

这个 GrpcRequest 是在我们自定义的拦截器中创建并传递的。

而我这里需要的请求包大小也是在拦截中获取到数据然后写入进 GrpcRequest。

static <T> Long getBodySize(T message) {  
  if (message instanceof MessageLite) {  
    return (long) ((MessageLite) message).getSerializedSize();  
  } else {  
    // Message is not a protobuf message  
    return null;  
  }}

这样就可以实现不同的 RPC 中获取自己的 attribute，同时每一组 attribute 也都是隔离的，互相解耦。

自定义 metrics

每个插件自定义 Metrics 的逻辑也是类似的，需要由框架层面提供 API 接口：

public InstrumenterBuilder<REQUEST, RESPONSE> addOperationMetrics(OperationMetrics factory) {  
  operationMetrics.add(requireNonNull(factory, "operationMetrics"));  
  return this;  
}
// 客户端的 metrics
.addOperationMetrics(RpcClientMetrics.get());

// 服务端的 metrics
.addOperationMetrics(RpcServerMetrics.get());

之后也会在框架层面回调这些自定义的 OperationMetrics:

    if (operationListeners.length != 0) {
      // operation listeners run after span start, so that they have access to the current span
      // for capturing exemplars
      long startNanos = getNanos(startTime);
      for (int i = 0; i < operationListeners.length; i++) {
        context = operationListeners[i].onStart(context, attributes, startNanos);
      }
    }

	if (operationListeners.length != 0) {  
	  long endNanos = getNanos(endTime);  
	  for (int i = operationListeners.length - 1; i >= 0; i--) {  
	    operationListeners[i].onEnd(context, attributes, endNanos);  
	  }
	}

这其中最关键的就是两个函数 onStart 和 onEnd，分别会在当前这个 span 的开始和结束时进行回调。

所以通常的做法是在 onStart 函数中初始化数据，然后在 onEnd 结束时统计结果，最终可以拿到 metrics 所需要的数据。

以这个 rpc.client.duration 客户端的请求耗时指标为例：

@Override  
public Context onStart(Context context, Attributes startAttributes, long startNanos) {  
  return context.with(  
      RPC_CLIENT_REQUEST_METRICS_STATE,  
      new AutoValue_RpcClientMetrics_State(startAttributes, startNanos));  
}

@Override  
public void onEnd(Context context, Attributes endAttributes, long endNanos) {  
  State state = context.get(RPC_CLIENT_REQUEST_METRICS_STATE);
	Attributes attributes = state.startAttributes().toBuilder().putAll(endAttributes).build();  
	clientDurationHistogram.record(  
	    (endNanos - state.startTimeNanos()) / NANOS_PER_MS, attributes, context);
}

在开始时记录下当前的时间，结束时获取当前时间和结束时间的差值正好就是这个 span 的执行时间，也就是 rpc client 的处理时间。

在 OpenTelemetry 中绝大多数的请求时间都是这么记录的。

Golang 增强

而在 Golang 中因为没有 byte-buddy 这种魔法库的存在，不可以直接修改源码，所以通常的做法还是得硬编码才行。

还是以 gRPC 为例，我们在创建 gRPC server 时就得指定一个 OpenTelemetry 提供的函数。

s := grpc.NewServer(  
    grpc.StatsHandler(otelgrpc.NewServerHandler()),  
)

在这个 SDK 中也会实现刚才在 Java 里类似的逻辑，限于篇幅具体逻辑就不细讲了。

总结

以上就是 gRPC 在 OpenTelemetry 中的具体实现，主要就是在找到需要增强框架是否有提供扩展的接口，如果有就直接使用该接口进行埋点。

如果没有那就需要查看源码，找到核心逻辑，再使用 byte-buddy 进行埋点。

比如 Pulsar 并没有在客户端提供一些扩展接口，只能找到它的核心函数进行埋点。

而在具体埋点过程中 OpenTelemetry 提供了许多解耦的 API，方便我们实现埋点所需要的业务逻辑，也会在后续的文章继续分析 OpenTelemetry 的一些设计原理和核心 API 的使用。

这部分 API 的设计我觉得是 OpenTelemetry 中最值得学习的地方。

参考链接：

标签：实战,context,gRPC,OpenTelemetry,grpc,io,public
From： https://www.cnblogs.com/crossoverJie/p/18395891

相关文章

  • OpenCV与AI深度学习 | 实战 | OpenCV传统方法实现密集圆形分割与计数(详细步骤 + 代码
    本文来源公众号“OpenCV与AI深度学习”，仅用于学术分享，侵权删，干货满满。原文链接：实战|OpenCV传统方法实现密集圆形分割与计数(详细步骤+代码)导 读    本文主要介绍基于OpenCV传统方法实现密集圆形分割与计数应用，并给详细步骤和代码。 背景介绍  实例图片来......
  • C#设计模式入门实战教程
    什么是设计模式设计模式是对面向对象设计中反复出现的问题的解决方案。它们提供了被反复使用、多数人知晓的、经过分类编目的代码设计经验总结。设计模式的作用提高代码的可重用性：通过定义一套标准的解决方案，设计模式使得相同或类似的问题可以在不同的项目中复用相同的代码结构或逻......
  • C#设计模式入门实战教程
    什么是设计模式设计模式是对面向对象设计中反复出现的问题的解决方案。它们提供了被反复使用、多数人知晓的、经过分类编目的代码设计经验总结。设计模式的作用提高代码的可重用性：通过定义一套标准的解决方案，设计模式使得相同或类似的问题可以在不同的项目中复用相同的代码结......
  • zdppy+vue3+onlyoffice文档管理系统实战 20240902 上课笔记 登录功能优化
    遗留问题1、登录以后跳转最近文档2、如果用户没有登录应该自动跳转登录页面3、如果用户的token校验失败，应该自动调整登录界面4、按回车键自动跳转登录页面登录以后跳转最近文档constrouter=useRouter()router.push("/")实际代码：constloginData=awaitapi.login......
  • FastAPI+Vue3零基础开发ERP系统项目实战课 20240831上课笔记 查询参数和分页实现
    回顾获取路径参数什么是路径参数？/user/{id}什么时候使用？需要传递参数怎么实现类型转换？声明参数的类型怎么捕获文件路径？{file_path:path}什么是查询参数查询字符串是键值对的集合，这些键值对位于URL的?之后，以&分隔。http://127.0.0.1:8000/items/?skip=0&limit=10......
  • 豆瓣评分9.7 这本书真是太优秀了! TensorFlow全面实战 附电子版！！
    前言TensorFlow乃是由GoogleBrain团队所开发的开源机器学习框架。其被广泛应用于各类机器学习与深度学习的领域，像是神经网络、自然语言处理以及图像识别等等。《TensorFlow实战：Google深度学习框架》全方位涵盖了TensorFlow的核心概念与技术，从基础的知识到高级的应用均有详......
  • 《BERT基础教程：Transformer大模型实战》一本相见恨晚的大模型入门书（附PDF）
    前言随着chatgpt的火热，大模型成为业界新的热点。而未来每个人，不仅仅是人工智能或者计算机行业从业者，未来的工作或多或少可能都会被大模型的影响。所以了解transformer为基础的大模型（Bert，gpt）就很有必要了。本书聚焦谷歌公司开发的BERT自然语言处理模型，由浅入深地介绍了BERT的工......
  • Vue+Express全栈开发项目实战技能：‌从0到1打造完整电商项目
    Vue+Express全栈开发项目实战技能：‌从0到1打造完整电商项目在Web开发领域，‌全栈开发是一个涵盖前端和后端技术的综合技能。‌本文将带你通过Vue.js和Express.js来实战开发一个完整的电商项目，‌从项目规划到部署上线，‌逐步掌握全栈开发的精髓。‌一、‌项目规划1.‌需求分析‌：‌......
  • 实战复杂低代码项目React从架构到拆解
    实战复杂低代码项目：‌React从架构到拆解随着技术的不断进步和业务需求的日益复杂，‌低代码平台（‌Low-CodePlatform）‌已成为现代软件开发领域中的热门工具。‌而基于React的低代码平台，‌更是凭借其组件化、‌响应式、‌数据驱动等特性，‌为开发者提供了一种高效构建应用程序的新途......
  • 攻破工程级复杂缓存难题--企业实战12
    缓存技术在现代分布式系统中至关重要，不仅提升了系统性能，还减轻了后端数据库的压力。然而，缓存系统也面临着诸多挑战，如缓存穿透、缓存雪崩、缓存击穿和热点key问题。通过多种策略的综合应用，包括本地缓存、双缓存方案、多级缓存、多副本、热点key拆分和动态分散等，可以有效应对这些......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99