OpenTelemetry与Reactive Java：如何为Mono添加返回值及执行状态标签

当然可行，你可以结合Mono的信号处理方法（doOnSuccess、doOnError）和Micrometer Observation的API，动态添加调用结果状态标签以及返回结果相关的标签。以下是两种具体实现方案：

方案1：基于现有Observation上下文动态添加标签 #

在doOnSuccess和doOnError回调中，通过Observation.getCurrent()获取当前追踪的Observation实例，动态补充标签信息：

@GetMapping(value = "/getClient")
public Mono<ResponseEntity<String>> getClient(HttpServletRequest request, final ClientRequest body) {
    return myService
            .getClient(body.getClient_id(), body.getProvider())
            .doOnError(throwable -> {
                log.error("Failed to get client", throwable);
                // 添加错误状态标签和错误类型标签
                Observation.getCurrent().ifPresent(obs -> {
                    obs.tag("status", "failure");
                    obs.tag("error_type", throwable.getClass().getSimpleName());
                });
            })
            .doOnSuccess(clientResult -> {
                // 添加成功状态标签和返回结果相关标签（示例：基于返回字符串长度）
                Observation.getCurrent().ifPresent(obs -> {
                    obs.tag("status", "success");
                    obs.tag("client_result_length", String.valueOf(clientResult.length()));
                    // 若返回是实体对象，可替换为具体字段，比如obs.tag("client_name", clientResult.getName());
                });
            })
            .name("get_client")
            .tag("client_id", body.getClient_id())
            .tag("provider", body.getProvider())
            .tap(Micrometer.observation(observationRegistry))
            .map(ResponseEntity::ok)
            .onErrorResume(e -> Mono.just(ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body("Error")));
}

方案2：手动创建Observation（更精细化控制） #

如果需要对观测的生命周期做更精细的管理，可以手动初始化Observation，在Mono的各个阶段手动启动、更新标签并结束观测：

@GetMapping(value = "/getClient")
public Mono<ResponseEntity<String>> getClient(HttpServletRequest request, final ClientRequest body) {
    // 初始化Observation并添加静态标签
    Observation observation = Observation.createNotStarted("get_client", observationRegistry)
            .tag("client_id", body.getClient_id())
            .tag("provider", body.getProvider());

    return myService
            .getClient(body.getClient_id(), body.getProvider())
            .doOnSubscribe(s -> observation.start()) // 订阅时启动观测
            .doOnError(throwable -> {
                log.error("Failed to get client", throwable);
                observation.tag("status", "failure")
                          .tag("error_type", throwable.getClass().getSimpleName())
                          .error(throwable) // 将错误绑定到观测
                          .stop(); // 结束观测
            })
            .doOnSuccess(clientResult -> {
                observation.tag("status", "success")
                          .tag("client_result_length", String.valueOf(clientResult.length()))
                          .stop(); // 结束观测
            })
            .map(ResponseEntity::ok)
            .onErrorResume(e -> Mono.just(ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body("Error")));
}

关键注意事项 #

• 状态标签：通过status标签明确区分调用成功/失败，失败时补充error_type标签可快速定位错误类型。
• 返回结果标签：仅提取有分析价值的字段（如关键标识、结果长度等），避免添加过大或敏感数据，防止观测数据冗余或信息泄露。
• 观测生命周期：手动创建Observation时，务必保证在订阅时启动、在信号完成/出错时结束，避免资源泄漏。

内容的提问来源于stack exchange，提问作者Stéphane Piette