Rust 网络编程进阶:集成 OpenTelemetry 实现全链路分布式追踪

在之前的程序中，我们已经集成了Prometheus，这让网关拥有了Metrics维度的监控能力。让我们可以了解到“当前的 QPS 是多少？”、“P99 延迟是多少？”、“错误率有没有超标？”这些聚合层面的问题。但是更深层面的跟踪问题比如某个请求的失败原因，请求耗时过长的原因，请求在微服务链路中到底经过了哪些节点，我们都无法跟踪到。今天集成的 OpenTelemetry将实现请求的全链路分布式追踪，并将追踪数据上报给 Jaeger，实现可视化追踪。

分布式追踪

核心原理

在微服务架构的服务中，一个请求会经过多个服务节点。分布式追踪就是要给每个请求打上一个唯一的 Trace ID。

网关的职责

网关处于链路的最前端需要承担生成 Trace ID以及注入 Trace ID的任务。

1. 生成：如果请求没有 Trace ID，则生成一个新的。
2. 注入：将 Trace ID和 Span ID按照 W3C 标准（traceparentHeader）注入到发往上游的 HTTP请求头中。

下游服务（无论是 Java、Go 还是 Node.js）只要遵循 OTel 标准，就能从 Header 中提取 Trace ID，并将自己的调用链串联起来。

升级改造

引入依赖

[dependencies]
opentelemetry = "0.31"
opentelemetry_sdk = { version = "0.31", features = ["rt-tokio"] }
opentelemetry-otlp = { version = "0.31", features = ["grpc-tonic"] }
tracing-opentelemetry = "0.32"
opentelemetry-http = "0.31"

tracing-subscriber = { version = "0.3", features = ["env-filter", "registry"] }

opentelemetry: 核心 API

opentelemetry_sdk: 需要开启 rt-tokio以支持异步运行时

opentelemetry-otlp: 需要开启 grpc-tonic，用来创建 Exporter并发送数据给 JaegerÏ

opentelemetry-http: 约定 HTTP语义，用于处理 Header

tracing-opentelemetry: 桥接 tracingcrate 的日志到 OTel

config.toml 新增配置

[server.tracing]
enabled = true
endpoint = "http://localhost:4317"

enabled: 是否开启分布式追踪，false 则使用原来的默认日志记录方式。

endpoint 是 Jaeger的 OTLP gRPC 端口

修改配置对应的结构体

#[derive(Debug, Deserialize, Clone)]
structServerConfig {
    listen_addr: String,
    cert_file: String,
    key_file: String,
    ip_limit: Option<RateLimitConfig>,
    tracing: Option<TracingConfig>,
}

#[derive(Debug, Deserialize, Clone)]
structTracingConfig {
    enabled: bool,
    endpoint: String,
}

ServerConfig下新增 tracing字段日志配置信息

初始化 Tracer

在 main函数启动时初始化全局 Tracer。

use tracing_subscriber::{Registry, layer::SubscriberExt, util::SubscriberInitExt};
use opentelemetry::{KeyValue, global};
use opentelemetry_otlp::WithExportConfig;
use opentelemetry_sdk::trace::SdkTracerProvider;
use opentelemetry_sdk::{propagation::TraceContextPropagator, resource::Resource};
use opentelemetry::trace::TracerProvider;
use tracing_opentelemetry::OpenTelemetrySpanExt;

fninit_tracer(config: &Option<TracingConfig>) {
    global::set_text_map_propagator(TraceContextPropagator::new());

ifletSome(conf) = config {
if !conf.enabled {
            init_default_logger();
return;
        }

let exporter = opentelemetry_otlp::SpanExporter::builder()
            .with_tonic()
            .with_endpoint(&conf.endpoint)
            .build()
            .expect("Failed to create OTLP exporter");

let resource = Resource::builder()
            .with_service_name("hyper-proxy-tool") // 设置服务名
            .with_attribute(KeyValue::new("service.version", "0.1.0")) // 设置版本号
            .build();

let provider = SdkTracerProvider::builder()
            .with_batch_exporter(exporter)
            .with_resource(resource)
            .build();

let tracer = provider.tracer("hyper-proxy-tool");
        global::set_tracer_provider(provider);

let telemetry = tracing_opentelemetry::layer().with_tracer(tracer);

let env_filter = tracing_subscriber::EnvFilter::try_from_default_env()
            .unwrap_or_else(|_| "hyper_proxy_tool=info".into());

let fmt_layer = tracing_subscriber::fmt::layer();

        Registry::default()
            .with(env_filter)
            .with(fmt_layer)
            .with(telemetry)
            .init();

        info!(
"🔭 Distributed Tracing enabled. Sending to {}",
            conf.endpoint
        );
    } else {
        init_default_logger();
    }
}

fninit_default_logger() {
let env_filter = tracing_subscriber::EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| "hyper_proxy_tool=info".into());
let fmt_layer = tracing_subscriber::fmt::layer();

    Registry::default().with(env_filter).with(fmt_layer).init();
}

global::set_text_map_propagator...: 使用 W3C Trace Context 标准设置全局 Propagator。在 traceparent 和 tracestate 头部中传播 SpanContext。

traceparent头部采用通用格式表示跟踪系统中的传入请求。格式为：traceparent: 00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01。

opentelemetry_otlp::SpanExporter: 使用 Builder模式创建 OTLP Exporter

Resource::builder(): 构建 Resource，用来定义服务标识（服务名和版本号）

SdkTracerProvider::builder(): 构建 SdkTracerProvider，传入 Exporter和 Resource

global::set_tracer_provider...: 获取 Tracer并设置全局 Provider，防止被 Drop

tracing_opentelemetry::layer().with_tracer(tracer);: 桥接 tracing， 这样 log/tracing 的 info!/error! 宏会自动关联 Trace ID

最终通过 tracing_subscriber::Registry注册订阅到 Subscriber中。

main 中调用 init_tracer

在 main开头初始化

let args = Cli::parse();    // 加载配置，获取 Tracing 配置
let config_path = args.config.clone();

let initial_config_str = fs::read_to_string(&config_path).unwrap_or_default();
let initial_config: Option<AppConfig> = toml::from_str(&initial_config_str).ok();

ifletSome(cfg) = &initial_config {
    init_tracer(&cfg.server.tracing); // 初始化 Tracing 
} else {
    init_default_logger();
}

Header 注入

在 proxy_handler发送请求前，把当前的 Trace Context注入到 HTTP Header。

定义一个辅助结构体来实现 Injectortrait

use hyper::header::HeaderName;
use opentelemetry::propagation::Injector;

structHeaderInjector<'a>(&'amut hyper::HeaderMap);

impl<'a> Injector for HeaderInjector<'a> {
fnset(&mutself, key: &str, value: String) {
ifletOk(name) = HeaderName::from_bytes(key.as_bytes()) {
ifletOk(val) = hyper::header::HeaderValue::from_str(&value) {
self.0.insert(name, val);
            }
        }
    }
}

在 proxy_handler中注入

// ... 在构建 Request Builder 时 ...

let ctx = tracing::Span::current().context();
ifletSome(headers) = builder.headers_mut() {
    global::get_text_map_propagator(|propagator| {
        propagator.inject_context(&ctx, &mut HeaderInjector(headers))
    });
}

// 以下是原来代码...
let retry_req = match builder.body(body_full.clone()) {
Ok(r) => r,
Err(_) => continue,
};

tracing::Span::current().context();: 获取当前的 Trace Context，由 tracing宏自动生成或从客户端请求继承

builder.headers_mut() {...}: 将 Trace ID注入到发往上游的 Header中

注意⚠️在缓冲模式和流式模式下都要注入

部署 Jaeger

使用 Docker Compose 启动 Jaeger。

docker-compose.yml:

services:
jaeger:
image:jaegertracing/all-in-one:latest
container_name:jaeger
ports:
-"16686:16686"# Web UI 面板
-"4317:4317"# OTLP gRPC 接收端口
-"4318:4318"# OTLP HTTP 接收端口
environment:
-COLLECTOR_OTLP_ENABLED=true

程序将通过 4317 端口发送数据到 jaeger 中，开头 config.toml中配置的 endpoint 就是指向 Jaeger

启动

docker-compose up -d

关闭

docker-compose down

验证测试

1. 启动网关程序和 Jaeger。
2. 发送请求：curl -k https://127.0.0.1:8443/api/v1/get。
3. 浏览器访问 Jaeger UI (http://localhost:16686)。

瀑布图如下：

可以看到 hyper-proxy-tool网关总耗时、http.method等详细信息。

日志传播：如果上游服务也集成了 OTel，上游的 Span 就会自动接在网关的 Span 下面，形成一个完整的调用链路。

完整代码

https://play.rust-lang.org/?version=stable&mode=debug&edition=2024&gist=82885b5485b751d7488b042d30cf5a19

总结

通过集成 OpenTelemetry，程序具备了“分布式全链路追踪”的能力，在观测性能以及跟踪 API 请求上下路由的能力提升了一个新的台阶。

Happy Coding with Rust🦀!

往期文章