OpenTelemetry 深度定制：跨服務追蹤的實戰技巧
背景
圖片
我們的需求是這樣的：
圖片
假設現在有三個服務：ServiceA、ServiceB、ServiceC。
ServiceA 對外提供了一個 http 接口 request，在這個接口會調用 ServiceB 的 order 訂單接口創建訂單，同時 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<HelloReply> 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<HelloReply> 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.extensinotallow=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 (ContextCustomizer<? super REQUEST> contextCustomizer : contextCustomizers) {
context = contextCustomizer.onStart(context, request, attributes);
}ContextCustomizer 是一個接口只提供了一個函數：
public interface ContextCustomizer<REQUEST> {
/** 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.
* * <p>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<GrpcRequest> {
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<REQUEST> build() {
Set<String> 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.extensinotallow=otel-extensions-custom-context-1.0-SNAPSHOT.jar \需要使用我們手動打包的 javaagent 以及一個自定義擴展包。
打包方式：
./gradlew assembleopentelemetry-java-instrumentation 項目比較大，所以打包過程可能比較久。
因為這其實是一些定制需求，所以就沒提交到上游，感興趣的可以自行合并代碼測試。
最后可以這個分支中查看到修改的部分：https://github.com/crossoverJie/opentelemetry-java-instrumentation/compare/main...add-grpc-context
