OpenTelemetry 實(shí)戰(zhàn)：gRPC 監(jiān)控的實(shí)現(xiàn)原理
前言
圖片
最近在給 opentelemetry-java-instrumentation 提交了一個(gè) PR，是關(guān)于給 gRPC 新增四個(gè) metrics：
- rpc.client.request.size: 客戶端請(qǐng)求包大小
- rpc.client.response.size：客戶端收到的響應(yīng)包大小
- rpc.server.request.size：服務(wù)端收到的請(qǐng)求包大小
- rpc.server.response.size：服務(wù)端響應(yīng)的請(qǐng)求包大小
這個(gè) PR 的主要目的就是能夠在指標(biāo)監(jiān)控中拿到 RPC 請(qǐng)求的包大小，而這里的關(guān)鍵就是如何才能拿到這些包的大小。
首先支持的是 gRPC（目前在云原生領(lǐng)域使用的最多），其余的 RPC 理論上也是可以支持的：
圖片
在實(shí)現(xiàn)的過程中我也比較好奇 OpenTelemetry 框架是如何給 gRPC 請(qǐng)求創(chuàng)建 span 調(diào)用鏈的，如下圖所示：
圖片
圖片
這是一個(gè) gRPC 遠(yuǎn)程調(diào)用，java-demo 是 gRPC 的客戶端，k8s-combat 是 gRPC 的服務(wù)端
在開始之前我們可以根據(jù) OpenTelemetry 的運(yùn)行原理大概猜測下它的實(shí)現(xiàn)過程。
首先我們應(yīng)用可以創(chuàng)建這些鏈路信息的前提是：使用了 OpenTelemetry 提供的 javaagent，這個(gè) agent 的原理是在運(yùn)行時(shí)使用了 byte-buddy 增強(qiáng)了我們應(yīng)用的字節(jié)碼，在這些字節(jié)碼中代理業(yè)務(wù)邏輯，從而可以在不影響業(yè)務(wù)的前提下增強(qiáng)我們的代碼（只要就是創(chuàng)建 span、metrics 等數(shù)據(jù)）
Spring 的一些代理邏輯也是這樣實(shí)現(xiàn)的
gRPC 增強(qiáng)原理
而在工程實(shí)現(xiàn)上，我們最好是不能對(duì)業(yè)務(wù)代碼進(jìn)行增強(qiáng)，而是要找到這些框架提供的擴(kuò)展接口。
拿 gRPC 來說，我們可以使用它所提供的 io.grpc.ClientInterceptor 和 io.grpc.ServerInterceptor 接口來增強(qiáng)代碼。
打開 io.opentelemetry.instrumentation.grpc.v1_6.TracingClientInterceptor 類我們可以看到它就是實(shí)現(xiàn)了 io.grpc.ClientInterceptor：
圖片
而其中最關(guān)鍵的就是要實(shí)現(xiàn) io.grpc.ClientInterceptor#interceptCall 函數(shù)：
@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);
}這個(gè)接口是 gRPC 提供的攔截器接口，對(duì)于 gRPC 客戶端來說就是在發(fā)起真正的網(wǎng)絡(luò)調(diào)用前后會(huì)執(zhí)行的方法。
所以在這個(gè)接口中我們就可以實(shí)現(xiàn)創(chuàng)建 span 獲取包大小等邏輯。
使用 byte-buddy 增強(qiáng)代碼
不過有一個(gè)問題是我們實(shí)現(xiàn)的 io.grpc.ClientInterceptor 類需要加入到攔截器中才可以使用：
var managedChannel = ManagedChannelBuilder.forAddress(host, port) .intercept(new TracingClientInterceptor()) // 加入攔截器
.usePlaintext()
.build();但在 javaagent 中是沒法給業(yè)務(wù)代碼中加上這樣的代碼的。
此時(shí)就需要 byte-buddy 登場了，它可以動(dòng)態(tài)修改字節(jié)碼從而實(shí)現(xiàn)類似于修改源碼的效果。
在 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>) 函數(shù)。
io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers#extendsClass/ isMethod 等函數(shù)都是 byte-buddy 庫提供的函數(shù)。
而這個(gè)函數(shù)正好就是我們需要在業(yè)務(wù)代碼里加入攔截器的地方。
interceptors.add(0, GrpcSingletons.CLIENT_INTERCEPTOR);
GrpcSingletons.CLIENT_INTERCEPTOR = new TracingClientInterceptor(clientInstrumenter, propagators);通過這行代碼可以手動(dòng)將 OpenTelemetry 里的 TracingClientInterceptor 加入到攔截器列表中，并且作為第一個(gè)攔截器。
而這里的：
extendsClass(named("io.grpc.ManagedChannelBuilder"))
.and(declaresField(named("interceptors")))通過函數(shù)的名稱也可以看出是為了找到 繼承了io.grpc.ManagedChannelBuilder 類中存在成員變量 interceptors 的類。
transformer.applyAdviceToMethod(
isMethod().and(named("build")),
GrpcClientBuilderBuildInstrumentation.class.getName() + "$AddInterceptorAdvice");然后在調(diào)用 build 函數(shù)后就會(huì)進(jìn)入自定義的 AddInterceptorAdvice 類，從而就可以攔截到添加攔截器的邏輯，然后把自定義的攔截器加入其中。
獲取 span 的 attribute
圖片
我們?cè)?gRPC 的鏈路中還可以看到這個(gè)請(qǐng)求的具體屬性，比如：
- gRPC 服務(wù)提供的 IP 端口。
- 請(qǐng)求的響應(yīng)碼
- 請(qǐng)求的 service 和 method
- 線程等信息。
這些信息在問題排查過程中都是至關(guān)重要的。
可以看到這里新的 attribute 主要是分為了三類：
- net.* 是網(wǎng)絡(luò)相關(guān)的屬性
- rpc.* 是和 grpc 相關(guān)的屬性
- thread.* 是線程相關(guān)的屬性
所以理論上我們?cè)谠O(shè)計(jì) API 時(shí)最好可以將這些不同分組的屬性解耦開，如果是 MQ 相關(guān)的可能還有一些 topic 等數(shù)據(jù)，所以各個(gè)屬性之間是互不影響的。
帶著這個(gè)思路我們來看看 gRPC 這里是如何實(shí)現(xiàn)的。
clientInstrumenterBuilder
.setSpanStatusExtractor(GrpcSpanStatusExtractor.CLIENT)
.addAttributesExtractors(additionalExtractors)
.addAttributesExtractor(RpcClientAttributesExtractor.create(rpcAttributesGetter))
.addAttributesExtractor(ServerAttributesExtractor.create(netClientAttributesGetter))
.addAttributesExtractor(NetworkAttributesExtractor.create(netClientAttributesGetter))OpenTelemetry 會(huì)提供一個(gè) io.opentelemetry.instrumentation.api.instrumenter.InstrumenterBuilder#addAttributesExtractor構(gòu)建器函數(shù)，用于存放自定義的屬性解析器。
從這里的源碼可以看出分別傳入了網(wǎng)絡(luò)相關(guān)、RPC 相關(guān)的解析器；正好也就對(duì)應(yīng)了圖中的那些屬性，也滿足了我們剛才提到的解耦特性。
而每一個(gè)自定義屬性解析器都需要實(shí)現(xiàn)接口 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，也就是對(duì)于到頁面上的 rpc.system 屬性。
而這里的 getService 函數(shù)則是拿來獲取 rpc.service 屬性的，可以看到它是通過 gRPC 的method 信息來獲取 service 的。
public interface RpcAttributesGetter<REQUEST> {
@Nullable
String getService(REQUEST request);
}而這里 REQUEST 其實(shí)是一個(gè)泛型，在 gRPC 里是 GrpcRequest，在其他 RPC 里這是對(duì)應(yīng)的 RPC 的數(shù)據(jù)。
這個(gè) GrpcRequest 是在我們自定義的攔截器中創(chuàng)建并傳遞的。
而我這里需要的請(qǐng)求包大小也是在攔截中獲取到數(shù)據(jù)然后寫入進(jìn) 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;
}}這樣就可以實(shí)現(xiàn)不同的 RPC 中獲取自己的 attribute，同時(shí)每一組 attribute 也都是隔離的，互相解耦。
自定義 metrics
每個(gè)插件自定義 Metrics 的邏輯也是類似的，需要由框架層面提供 API 接口：
public InstrumenterBuilder<REQUEST, RESPONSE> addOperationMetrics(OperationMetrics factory) {
operationMetrics.add(requireNonNull(factory, "operationMetrics"));
return this;
}
// 客戶端的 metrics
.addOperationMetrics(RpcClientMetrics.get());
// 服務(wù)端的 metrics
.addOperationMetrics(RpcServerMetrics.get());之后也會(huì)在框架層面回調(diào)這些自定義的 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);
}
}這其中最關(guān)鍵的就是兩個(gè)函數(shù) onStart 和 onEnd，分別會(huì)在當(dāng)前這個(gè) span 的開始和結(jié)束時(shí)進(jìn)行回調(diào)。
所以通常的做法是在 onStart 函數(shù)中初始化數(shù)據(jù)，然后在 onEnd 結(jié)束時(shí)統(tǒng)計(jì)結(jié)果，最終可以拿到 metrics 所需要的數(shù)據(jù)。
以這個(gè) rpc.client.duration 客戶端的請(qǐng)求耗時(shí)指標(biāo)為例：
@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);
}在開始時(shí)記錄下當(dāng)前的時(shí)間，結(jié)束時(shí)獲取當(dāng)前時(shí)間和結(jié)束時(shí)間的差值正好就是這個(gè) span 的執(zhí)行時(shí)間，也就是 rpc client 的處理時(shí)間。
在 OpenTelemetry 中絕大多數(shù)的請(qǐng)求時(shí)間都是這么記錄的。
Golang 增強(qiáng)
而在 Golang 中因?yàn)闆]有 byte-buddy 這種魔法庫的存在，不可以直接修改源碼，所以通常的做法還是得硬編碼才行。
還是以 gRPC 為例，我們?cè)趧?chuàng)建 gRPC server 時(shí)就得指定一個(gè) OpenTelemetry 提供的函數(shù)。
s := grpc.NewServer(
grpc.StatsHandler(otelgrpc.NewServerHandler()),
)
在這個(gè) SDK 中也會(huì)實(shí)現(xiàn)剛才在 Java 里類似的邏輯，限于篇幅具體邏輯就不細(xì)講了。
總結(jié)
以上就是 gRPC 在 OpenTelemetry 中的具體實(shí)現(xiàn)，主要就是在找到需要增強(qiáng)框架是否有提供擴(kuò)展的接口，如果有就直接使用該接口進(jìn)行埋點(diǎn)。
如果沒有那就需要查看源碼，找到核心邏輯，再使用 byte-buddy 進(jìn)行埋點(diǎn)。
圖片
比如 Pulsar 并沒有在客戶端提供一些擴(kuò)展接口，只能找到它的核心函數(shù)進(jìn)行埋點(diǎn)。
而在具體埋點(diǎn)過程中 OpenTelemetry 提供了許多解耦的 API，方便我們實(shí)現(xiàn)埋點(diǎn)所需要的業(yè)務(wù)邏輯，也會(huì)在后續(xù)的文章繼續(xù)分析 OpenTelemetry 的一些設(shè)計(jì)原理和核心 API 的使用。
這部分 API 的設(shè)計(jì)我覺得是 OpenTelemetry 中最值得學(xué)習(xí)的地方。
參考鏈接：
- https://bytebuddy.net/#/
- https://opentelemetry.io/docs/specs/semconv/rpc/rpc-metrics/#metric-rpcserverrequestsize
