tdm_server_rust/telemetry/
layers.rs

1//! tracing 订阅器 Layer 组装与日志 trace_id 格式
2
3use crate::{
4    config::TelemetryConfig,
5    profile::DevProfileLayer,
6    telemetry::{logs, otlp, skywalking},
7};
8use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
9use std::path::Path;
10use std::sync::Arc;
11use tracing::Level;
12use tracing::Subscriber;
13use tracing_subscriber::{
14    fmt::{format::Writer, time::FormatTime, FormatEvent, FormatFields},
15    layer::SubscriberExt,
16    util::SubscriberInitExt,
17    EnvFilter, Layer, Registry,
18};
19
20/// 可观测性守卫：持有 SW reporter、OTel Logger 与文件 appender
21pub struct TelemetryGuard {
22    /// OTel Log Provider
23    providers: Option<otlp::OtelProviders>,
24    /// SkyWalking gRPC 上报
25    skywalking: Option<skywalking::SkyWalkingHandle>,
26    /// 异步日志落盘 guard（pro）
27    _file_guard: Option<tracing_appender::non_blocking::WorkerGuard>,
28}
29
30impl TelemetryGuard {
31    /// 进程退出前 flush
32    pub fn shutdown(self) {
33        if let Some(p) = self.providers {
34            if let Some(tp) = p.tracer_provider {
35                let _ = tp.shutdown();
36            }
37            let _ = p.logger_provider.shutdown();
38        }
39    }
40
41    /// SW Logger（debug_log / error_log 结构化上报）
42    pub fn sw_logger(&self) -> Option<Arc<::skywalking::logging::logger::Logger>> {
43        self.skywalking.as_ref().map(|h| h.logger.clone())
44    }
45}
46
47/// 初始化 tracing + SW 原生 + OTLP Log
48pub async fn init(profile: &str, cfg: &TelemetryConfig) -> TelemetryGuard {
49    let is_dev = matches!(profile, "dev" | "dev-h2");
50    let filter = build_env_filter(cfg);
51
52    let skywalking = if cfg.enabled && cfg.skywalking.export_traces {
53        match skywalking::init_skywalking(cfg).await {
54            Ok(h) => Some(h),
55            Err(e) => {
56                eprintln!("SkyWalking 初始化失败，继续无 Segment 导出: {e}");
57                None
58            }
59        }
60    } else {
61        None
62    };
63
64    let providers = if cfg.enabled {
65        match otlp::build_providers(profile, cfg) {
66            Ok(p) => Some(p),
67            Err(e) => {
68                eprintln!("OTel Log 初始化失败，继续无 OTLP 日志: {e}");
69                None
70            }
71        }
72    } else {
73        None
74    };
75
76    let mut file_guard = None;
77    let mut fmt_layer = tracing_subscriber::fmt::layer()
78        .with_target(false)
79        .with_ansi(is_dev)
80        .event_format(OtelLogFormat)
81        .boxed();
82
83    if !is_dev && cfg.log_file.non_blocking {
84        let path = Path::new(&cfg.log_file.path);
85        if let Some(dir) = path.parent() {
86            let _ = std::fs::create_dir_all(dir);
87        }
88        let file_name = path
89            .file_name()
90            .and_then(|s| s.to_str())
91            .unwrap_or("app.log");
92        let dir = path
93            .parent()
94            .map(|p| p.to_string_lossy().into_owned())
95            .unwrap_or_else(|| "./logs".into());
96        let file_appender = tracing_appender::rolling::daily(&dir, file_name);
97        let (non_blocking, guard) = tracing_appender::non_blocking(file_appender);
98        file_guard = Some(guard);
99        fmt_layer = tracing_subscriber::fmt::layer()
100            .with_target(false)
101            .with_ansi(false)
102            .event_format(OtelLogFormat)
103            .with_writer(non_blocking)
104            .boxed();
105    }
106
107    let sw_log_layer: skywalking::SwLogLayer = skywalking
108        .as_ref()
109        .filter(|_| cfg.skywalking.export_native_logs)
110        .map(|h| {
111            let min = parse_level(&cfg.log_level);
112            skywalking::SwLogLayer::Active(skywalking::SkyWalkingLogLayer::new(
113                h.logger.clone(),
114                min,
115            ))
116        })
117        .unwrap_or(skywalking::SwLogLayer::noop());
118
119    let registry = Registry::default()
120        .with(filter)
121        .with(fmt_layer)
122        .with(sw_log_layer);
123
124    if let Some(ref p) = providers {
125        match (logs::export_enabled(cfg), is_dev) {
126            (true, true) => registry
127                .with(OpenTelemetryTracingBridge::new(&p.logger_provider))
128                .with(DevProfileLayer::new())
129                .init(),
130            (true, false) => registry
131                .with(OpenTelemetryTracingBridge::new(&p.logger_provider))
132                .init(),
133            (false, true) => registry.with(DevProfileLayer::new()).init(),
134            (false, false) => registry.init(),
135        }
136    } else if is_dev {
137        registry.with(DevProfileLayer::new()).init();
138    } else {
139        registry.init();
140    }
141
142    TelemetryGuard {
143        providers,
144        skywalking,
145        _file_guard: file_guard,
146    }
147}
148
149fn parse_level(s: &str) -> Level {
150    match s.to_lowercase().as_str() {
151        "trace" => Level::TRACE,
152        "debug" => Level::DEBUG,
153        "warn" => Level::WARN,
154        "error" => Level::ERROR,
155        _ => Level::INFO,
156    }
157}
158
159/// 构建 EnvFilter
160fn build_env_filter(cfg: &TelemetryConfig) -> EnvFilter {
161    let default = format!(
162        "tdm_server_rust={},tower_http=warn,sqlx=warn",
163        cfg.log_level
164    );
165    EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(default))
166}
167
168/// 日志格式：输出 SW trace_id
169struct OtelLogFormat;
170
171impl<S, N> FormatEvent<S, N> for OtelLogFormat
172where
173    S: Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>,
174    N: for<'a> FormatFields<'a> + 'static,
175{
176    fn format_event(
177        &self,
178        ctx: &tracing_subscriber::fmt::FmtContext<'_, S, N>,
179        mut writer: Writer<'_>,
180        event: &tracing::Event<'_>,
181    ) -> std::fmt::Result {
182        use tracing_subscriber::fmt::time::SystemTime;
183
184        let fmt_time = SystemTime;
185        fmt_time.format_time(&mut writer)?;
186
187        write!(writer, " ")?;
188        let level = *event.metadata().level();
189        write!(writer, "{level:>5} ")?;
190
191        if let Some(trace_id) = skywalking::current_trace_id() {
192            write!(writer, "trace_id={trace_id} ")?;
193        }
194
195        ctx.field_format().format_fields(writer.by_ref(), event)?;
196        writeln!(writer)
197    }
198}
tdm_server_rust/telemetry/layers.rs

tdm_server_rust/telemetry/
layers.rs
