refactor: unify telemetry on the oltp crate and drop export telemetry

Modified app/src/main.rs +39 −11 View app/src/main.rs

@@ -30,7 +30,7 @@

3030 use utoipa_swagger_ui::SwaggerUi;

3131 use webapi::openapi::api_doc;

3232

33 export::telemetry::init_tracing();

33 let telemetry = oltp::init(&oltp::TelemetryConfig::from_env("gitdata"));

3434

3535 let cfg = ApihostConfig::load().context("loading configuration")?;

3636 if !cfg.apihost.enabled() {

@@ -85,10 +85,7 @@

8585 }

8686 info!("dependencies ready");

8787

88 if let Err(err) = export::telemetry::init_meter_provider() {

89 warn!(error = %err, "failed to init meter provider; metrics disabled");

90 }

91 export::metrics::spawn_queue_depth_reporter(

88 spawn_queue_depth_reporter(

9289 state.redis.clone(),

9390 &[

9491 ("push.jobs", export::push::QUEUE_TOPIC.to_string()),

@@ -139,7 +136,7 @@

139136 .wrap(auth::envelope::EnvelopeContext)

140137 .wrap(webapi::build_cors(&state.config.web))

141138 .wrap(from_fn(export::middleware::trace_request))

142 .wrap(from_fn(export::telemetry::http_telemetry))

139 .wrap(oltp::actix::HttpTelemetry::new("gitdata"))

143140 .wrap(from_fn(auth::share::session::revalidate::revalidate))

144141 .wrap(from_fn(auth::share::session::csrf::csrf))

145142 .service(web::resource("/api/docs").route(web::get().to(|| async {

@@ -165,9 +162,10 @@

165162 });

166163 let api = async move { server.await.context("actix server error") };

167164

165 let shutdown = oltp::shutdown::install();

168166 let signal = async {

169167 tokio::select! {

170 _ = export::telemetry::shutdown_signal() => cancel.cancel(),

168 _ = shutdown.wait() => cancel.cancel(),

171169 _ = cancel.cancelled() => {},

172170 }

173171 };

@@ -181,7 +179,9 @@

181179 conn.shutdown().await;

182180 api_result?;

183181

184 export::telemetry::shutdown_telemetry();

182 if let Err(error) = telemetry.shutdown() {

183 warn!(%error, "flushing telemetry failed");

184 }

185185

186186 info!("apihost shut down");

187187 Ok(())

@@ -191,7 +191,7 @@

191191async fn mail() -> anyhow::Result<()> {

192192 use tokio_util::sync::CancellationToken;

193193

194 export::telemetry::init_tracing();

194 let telemetry = oltp::init(&oltp::TelemetryConfig::from_env("gitdata-mail"));

195195

196196 let figment = config::loader::figment().context("loading configuration")?;

197197 let rocketmq: config::rocketmq::RocketMqConfig = figment

@@ -217,16 +217,19 @@

217217 let cancel = CancellationToken::new();

218218 let run_cancel = cancel.clone();

219219 let run = consumer.run(rocketmq, run_cancel);

220 let shutdown = oltp::shutdown::install();

220221 let signal = async {

221222 tokio::select! {

222 _ = export::telemetry::shutdown_signal() => cancel.cancel(),

223 _ = shutdown.wait() => cancel.cancel(),

223224 _ = cancel.cancelled() => {},

224225 }

225226 };

226227 let (result, ()) = tokio::join!(run, signal);

227228 conn.shutdown().await;

228229 result.context("mail service failed")?;

229 export::telemetry::shutdown_telemetry();

230 if let Err(error) = telemetry.shutdown() {

231 tracing::warn!(%error, "flushing telemetry failed");

232 }

230233 Ok(())

231234}

232235

@@ -339,3 +342,28 @@

339342 println!("wrote {}", path.display());

340343 Ok(())

341344}

345

346const QUEUE_DEPTH_INTERVAL: std::time::Duration = std::time::Duration::from_secs(30);

347

348/// 周期性上报队列与死信队列的积压深度；进程退出时随运行时一起结束。

349fn spawn_queue_depth_reporter(redis: export::AppRedis, queues: &[(&'static str, String)]) {

350 if queues.is_empty() {

351 return;

352 }

353 let queues = queues.to_vec();

354 tokio::spawn(async move {

355 let mut ticker = tokio::time::interval(QUEUE_DEPTH_INTERVAL);

356 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

357 loop {

358 ticker.tick().await;

359 for (topic, key) in &queues {

360 match redis.llen(key).await {

361 Ok(depth) => oltp::metrics::queue_depth(topic, depth),

362 Err(error) => {

363 tracing::debug!(%error, %key, "queue depth sample failed");

364 }

365 }

366 }

367 }

368 });

369}

Modified app/Cargo.toml +1 −0 View app/Cargo.toml

@@ -17,6 +17,7 @@

1717export = { workspace = true }

1818session = { workspace = true }

1919models = { workspace = true }

20oltp = { workspace = true }

2021push = { workspace = true }

2122queue = { workspace = true }

2223sea-orm = { workspace = true }

Modified lib/chat/src/services/messages.rs +2 −2 View lib/chat/src/services/messages.rs

@@ -100,8 +100,8 @@

100100 })),

101101 actor_id: Some(actor),

102102 notification_id: None,

103 trace: export::telemetry::traceparent(),

104 request_id: export::telemetry::request_id(),

103 trace: oltp::propagation::traceparent(),

104 request_id: export::middleware::request_id(),

105105 },

106106 )

107107 .await;

Modified lib/chat/src/services/outbound.rs +5 −5 View lib/chat/src/services/outbound.rs

@@ -371,8 +371,8 @@

371371 }

372372 let client = client_builder.build().map_err(|e| e.to_string())?;

373373

374 let span = export::telemetry::http_client_span("POST", &webhook.url);

375 let response = export::telemetry::send(

374 let span = oltp::client::http_client_span("POST", &webhook.url);

375 let response = oltp::client::send(

376376 span,

377377 client

378378 .post(&webhook.url)

@@ -384,15 +384,15 @@

384384 )

385385 .await

386386 .map_err(|e| {

387 export::metrics::chat_webhook_delivery("error");

387 oltp::metrics::chat_webhook_delivery("error");

388388 format!("request failed: {e}")

389389 })?;

390390 let status = response.status().as_u16();

391391 if (200..300).contains(&status) {

392 export::metrics::chat_webhook_delivery("ok");

392 oltp::metrics::chat_webhook_delivery("ok");

393393 Ok(format!("delivered {status}"))

394394 } else {

395 export::metrics::chat_webhook_delivery("error");

395 oltp::metrics::chat_webhook_delivery("error");

396396 Err(format!("endpoint returned {status}"))

397397 }

398398 }

Modified lib/chat/Cargo.toml +1 −0 View lib/chat/Cargo.toml

@@ -11,6 +11,7 @@

1111i18n = { workspace = true }

1212imaging = { workspace = true }

1313models = { workspace = true }

14oltp = { workspace = true }

1415queue = { workspace = true }

1516social = { workspace = true }

1617session = { workspace = true }

Modified lib/export/src/lib.rs +0 −2 View lib/export/src/lib.rs

@@ -6,7 +6,6 @@

66pub mod http;

77pub mod identity_client;

88pub mod imaging;

9pub mod metrics;

109pub mod middleware;

1110pub mod notify;

1211pub mod pagination;

@@ -15,7 +14,6 @@

1514pub mod rest;

1615pub mod serde_datetime;

1716pub mod serde_patch;

18pub mod telemetry;

1917pub mod urls;

2018pub mod users;

2119

Deleted lib/export/src/metrics.rs +0 −125

@@ -1,125 +0,0 @@

1use std::sync::LazyLock;

2use std::time::Duration;

3

4use opentelemetry::metrics::{Counter, Gauge, Meter, UpDownCounter};

5use opentelemetry::{KeyValue, global};

6

7use crate::redis::AppRedis;

8

9fn meter() -> Meter {

10 global::meter("gitdata")

11}

12

13static AUTH_LOGINS: LazyLock<Counter<u64>> = LazyLock::new(|| {

14 meter()

15 .u64_counter("auth.logins")

16 .with_description("Login attempts by outcome")

17 .build()

18});

19

20static GIT_OPS: LazyLock<Counter<u64>> = LazyLock::new(|| {

21 meter()

22 .u64_counter("git.ops")

23 .with_description("Git protocol operations by operation and outcome")

24 .build()

25});

26

27static PUSH_DELIVERIES: LazyLock<Counter<u64>> = LazyLock::new(|| {

28 meter()

29 .u64_counter("push.deliveries")

30 .with_description("Web push deliveries by outcome")

31 .build()

32});

33

34static IM_CONNECTIONS: LazyLock<UpDownCounter<i64>> = LazyLock::new(|| {

35 meter()

36 .i64_up_down_counter("im.connections")

37 .with_description("Open chat WebSocket connections")

38 .build()

39});

40

41static CHAT_WEBHOOK_DELIVERIES: LazyLock<Counter<u64>> = LazyLock::new(|| {

42 meter()

43 .u64_counter("chat.webhook.deliveries")

44 .with_description("Outbound chat webhook deliveries by outcome")

45 .build()

46});

47

48static QUEUE_DEPTH: LazyLock<Gauge<i64>> = LazyLock::new(|| {

49 meter()

50 .i64_gauge("queue.depth")

51 .with_description("Pending queue entries by topic")

52 .build()

53});

54

55pub fn auth_login(outcome: &'static str) {

56 AUTH_LOGINS.add(1, &[KeyValue::new("outcome", outcome)]);

57}

58

59pub fn git_op(op: &'static str, outcome: &'static str) {

60 GIT_OPS.add(

61 1,

62 &[KeyValue::new("op", op), KeyValue::new("outcome", outcome)],

63 );

64}

65

66pub fn push_delivery(outcome: &'static str) {

67 PUSH_DELIVERIES.add(1, &[KeyValue::new("outcome", outcome)]);

68}

69

70pub fn im_connection(delta: i64) {

71 IM_CONNECTIONS.add(delta, &[]);

72}

73

74/// `Drop`-based guard so every socket exit path (incl. early returns) balances

75/// the connection gauge.

76pub struct ImConnectionGuard;

77

78impl ImConnectionGuard {

79 #[must_use]

80 pub fn new() -> Self {

81 im_connection(1);

82 Self

83 }

84}

85

86impl Default for ImConnectionGuard {

87 fn default() -> Self {

88 Self::new()

89 }

90}

91

92impl Drop for ImConnectionGuard {

93 fn drop(&mut self) {

94 im_connection(-1);

95 }

96}

97

98pub fn chat_webhook_delivery(outcome: &'static str) {

99 CHAT_WEBHOOK_DELIVERIES.add(1, &[KeyValue::new("outcome", outcome)]);

100}

101

102const QUEUE_DEPTH_INTERVAL: Duration = Duration::from_secs(30);

103

104/// 周期性上报队列与死信队列的积压深度；进程退出时随运行时一起结束。

105pub fn spawn_queue_depth_reporter(redis: AppRedis, queues: &[(&'static str, String)]) {

106 let queues: Vec<(&'static str, String)> = queues.to_vec();

107 if queues.is_empty() {

108 return;

109 }

110 tokio::spawn(async move {

111 let mut ticker = tokio::time::interval(QUEUE_DEPTH_INTERVAL);

112 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

113 loop {

114 ticker.tick().await;

115 for (topic, key) in &queues {

116 match redis.llen(key).await {

117 Ok(depth) => QUEUE_DEPTH.record(depth, &[KeyValue::new("topic", *topic)]),

118 Err(error) => {

119 tracing::debug!(%error, %key, "queue depth sample failed");

120 }

121 }

122 }

123 }

124 });

125}

Modified lib/export/src/push.rs +2 −2 View lib/export/src/push.rs

@@ -136,8 +136,8 @@

136136 data,

137137 actor_id,

138138 notification_id,

139 trace: crate::telemetry::traceparent(),

140 request_id: crate::telemetry::request_id(),

139 trace: oltp::propagation::traceparent(),

140 request_id: crate::middleware::request_id(),

141141 })

142142 }

143143

Deleted lib/export/src/telemetry.rs +0 −400

@@ -1,400 +0,0 @@

1use std::sync::OnceLock;

2use std::time::Duration;

3

4use actix_web::body::{BoxBody, MessageBody};

5use actix_web::dev::{ServiceRequest, ServiceResponse};

6use actix_web::middleware::Next;

7use actix_web::{Error, HttpMessage, HttpRequest, HttpResponse};

8use http::{HeaderMap, Request, Response};

9use opentelemetry::propagation::TextMapCompositePropagator;

10use opentelemetry::trace::TracerProvider as _;

11use opentelemetry::{Context, global};

12use opentelemetry_http::{HeaderExtractor, HeaderInjector};

13use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider, Temporality};

14use opentelemetry_sdk::propagation::{BaggagePropagator, TraceContextPropagator};

15use opentelemetry_sdk::trace::SdkTracerProvider;

16use tracing::{Instrument as _, Span, info, warn};

17use tracing_opentelemetry::OpenTelemetrySpanExt;

18use tracing_subscriber::EnvFilter;

19use tracing_subscriber::layer::SubscriberExt;

20use tracing_subscriber::util::SubscriberInitExt;

21

22/// Installed tracer provider, kept so shutdown can flush the batch processor.

23static TRACER_PROVIDER: OnceLock<SdkTracerProvider> = OnceLock::new();

24/// Installed meter provider, kept so shutdown can flush the periodic reader.

25static METER_PROVIDER: OnceLock<SdkMeterProvider> = OnceLock::new();

26

27const METRICS_EXPORT_INTERVAL: Duration = Duration::from_secs(30);

28

29pub fn init_tracing() {

30 // W3C trace context is what lets `traceparent` survive HTTP, gRPC and

31 // queue payloads across hosts; installing it is a no-op for log-only runs.

32 global::set_text_map_propagator(TextMapCompositePropagator::new(vec![

33 Box::new(TraceContextPropagator::new()),

34 Box::new(BaggagePropagator::new()),

35 ]));

36

37 // The filter is installed on the registry rather than on the fmt layer so it

38 // governs span *creation*: without it every debug-level span in the process

39 // (hyper/tower internals) becomes an exported span.

40 let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));

41 let subscriber = tracing_subscriber::registry()

42 .with(tracing_subscriber::fmt::layer())

43 .with(filter);

44

45 match tracer_provider() {

46 Some(provider) => {

47 let _ = TRACER_PROVIDER.set(provider.clone());

48 subscriber

49 .with(tracing_opentelemetry::layer().with_tracer(provider.tracer("gitdata")))

50 .init();

51 info!("opentelemetry trace export enabled");

52 }

53 None => subscriber.init(),

54 }

55}

56

57pub fn init_meter_provider() -> Result<SdkMeterProvider, String> {

58 if !otlp_configured(OtlpSignal::Metrics) {

59 return Err("OTLP endpoint is not configured".to_string());

60 }

61

62 let exporter = opentelemetry_otlp::MetricExporter::builder()

63 .with_http()

64 .with_temporality(Temporality::default())

65 .build()

66 .map_err(|error| error.to_string())?;

67

68 let reader = PeriodicReader::builder(exporter)

69 .with_interval(METRICS_EXPORT_INTERVAL)

70 .build();

71

72 let provider = SdkMeterProvider::builder().with_reader(reader).build();

73 global::set_meter_provider(provider.clone());

74 let _ = METER_PROVIDER.set(provider.clone());

75 Ok(provider)

76}

77

78#[derive(Clone, Copy)]

79enum OtlpSignal {

80 Traces,

81 Metrics,

82}

83

84/// Telemetry is opt-in through the standard OTLP environment variables:

85/// setting `OTEL_EXPORTER_OTLP_ENDPOINT` turns export on, anything else

86/// (including a bad endpoint) only downgrades to log-only. Telemetry must never

87/// stop a host from booting, so every failure path here logs and returns `None`.

88fn otlp_configured(signal: OtlpSignal) -> bool {

89 let specific = match signal {

90 OtlpSignal::Traces => "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",

91 OtlpSignal::Metrics => "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",

92 };

93 ["OTEL_EXPORTER_OTLP_ENDPOINT", specific].iter().any(|key| {

94 std::env::var(key)

95 .map(|value| !value.trim().is_empty())

96 .unwrap_or(false)

97 })

98}

99

100fn tracer_provider() -> Option<SdkTracerProvider> {

101 if !otlp_configured(OtlpSignal::Traces) {

102 return None;

103 }

104

105 let exporter = match opentelemetry_otlp::SpanExporter::builder()

106 .with_http()

107 .build()

108 {

109 Ok(exporter) => exporter,

110 Err(err) => {

111 eprintln!("failed to build otlp span exporter, traces disabled: {err}");

112 return None;

113 }

114 };

115

116 Some(

117 SdkTracerProvider::builder()

118 .with_batch_exporter(exporter)

119 .build(),

120 )

121}

122

123/// Adopts the W3C trace context carried by inbound headers, when present.

124pub fn set_remote_parent(span: &Span, headers: &HeaderMap) {

125 let parent = global::get_text_map_propagator(|propagator| {

126 propagator.extract_with_context(&Context::new(), &HeaderExtractor(headers))

127 });

128 let _ = span.set_parent(parent);

129}

130

131/// Continues a trace across a queue boundary; `None` keeps the ambient parent.

132pub fn set_parent_from_traceparent(span: &Span, traceparent: Option<&str>) {

133 let Some(traceparent) = traceparent else {

134 return;

135 };

136 let mut headers = HeaderMap::new();

137 if let Ok(value) = traceparent.parse() {

138 headers.insert("traceparent", value);

139 }

140 set_remote_parent(span, &headers);

141}

142

143/// Serialises the current trace context for a queue payload.

144pub fn traceparent() -> Option<String> {

145 let mut headers = HeaderMap::new();

146 global::get_text_map_propagator(|propagator| {

147 propagator.inject_context(

148 &Span::current().context(),

149 &mut HeaderInjector(&mut headers),

150 );

151 });

152 headers

153 .get("traceparent")

154 .and_then(|value| value.to_str().ok())

155 .map(str::to_owned)

156}

157

158/// Serialises the request id for a queue payload.

159pub fn request_id() -> Option<String> {

160 crate::middleware::request_id()

161}

162

163fn span_with(

164 method: &str,

165 path: &str,

166 route: &str,

167 request_id: Option<&str>,

168 headers: &HeaderMap,

169) -> Span {

170 let name = format!("{method} {route}");

171 let span = tracing::info_span!(

172 "http.server",

173 "otel.name" = %name,

174 "otel.kind" = "server",

175 "http.request.method" = %method,

176 "url.path" = %path,

177 "http.route" = %route,

178 "http.response.status_code" = tracing::field::Empty,

179 "latency_ms" = tracing::field::Empty,

180 "request_id" = tracing::field::Empty,

181 );

182 if let Some(request_id) = request_id {

183 span.record("request_id", request_id);

184 }

185 set_remote_parent(&span, headers);

186 span

187}

188

189pub fn http_span_actix(request: &HttpRequest) -> Span {

190 let route = request

191 .match_pattern()

192 .unwrap_or_else(|| request.path().to_string());

193 let request_id = request

194 .extensions()

195 .get::<crate::middleware::RequestId>()

196 .map(|id| id.0.clone());

197 let mut headers = HeaderMap::new();

198 for (name, value) in request.headers() {

199 if let (Ok(name), Ok(value)) = (

200 http::header::HeaderName::from_bytes(name.as_str().as_bytes()),

201 http::HeaderValue::from_bytes(value.as_bytes()),

202 ) {

203 headers.insert(name, value);

204 }

205 }

206 span_with(

207 request.method().as_str(),

208 request.path(),

209 &route,

210 request_id.as_deref(),

211 &headers,

212 )

213}

214

215pub fn http_span<B>(request: &Request<B>) -> Span {

216 let path = request.uri().path();

217 let request_id = request

218 .extensions()

219 .get::<crate::middleware::RequestId>()

220 .map(|id| id.0.clone());

221 span_with(

222 request.method().as_str(),

223 path,

224 path,

225 request_id.as_deref(),

226 request.headers(),

227 )

228}

229

230pub fn on_response<B>(response: &HttpResponse<B>, latency: Duration, span: &Span) {

231 let status = response.status().as_u16();

232 span.record("http.response.status_code", i64::from(status));

233 span.record("latency_ms", latency.as_millis() as i64);

234 if status >= 500 {

235 span.record("otel.status_code", "error");

236 span.record("error.type", i64::from(status));

237 }

238}

239

240pub async fn http_telemetry<B>(

241 request: ServiceRequest,

242 next: Next<B>,

243) -> Result<ServiceResponse<BoxBody>, Error>

244where

245 B: MessageBody + 'static,

246{

247 let span = http_span_actix(request.request());

248 let started = std::time::Instant::now();

249 let response = next.call(request).instrument(span.clone()).await?;

250 let latency = started.elapsed();

251 on_response(response.response(), latency, &span);

252 tracing::info!(

253 parent: &span,

254 status = response.status().as_u16(),

255 latency_ms = latency.as_millis() as u64,

256 "finished processing request",

257 );

258 Ok(response.map_into_boxed_body())

259}

260

261/// Body-agnostic tower layer for servers whose body type must stay untouched

262/// (tonic gRPC): it adopts the remote parent, creates the server span and runs

263/// the inner future inside it.

264#[derive(Clone, Copy, Default)]

265pub struct TraceContextLayer;

266

267impl<S> tower::Layer<S> for TraceContextLayer {

268 type Service = TraceContextService<S>;

269

270 fn layer(&self, inner: S) -> Self::Service {

271 TraceContextService { inner }

272 }

273}

274

275#[derive(Clone)]

276pub struct TraceContextService<S> {

277 inner: S,

278}

279

280impl<S, B, ResBody> tower::Service<Request<B>> for TraceContextService<S>

281where

282 S: tower::Service<Request<B>, Response = Response<ResBody>>,

283 S::Future: Send + 'static,

284 B: 'static,

285{

286 type Response = S::Response;

287 type Error = S::Error;

288 type Future = tracing::instrument::Instrumented<S::Future>;

289

290 fn poll_ready(

291 &mut self,

292 cx: &mut std::task::Context<'_>,

293 ) -> std::task::Poll<Result<(), Self::Error>> {

294 self.inner.poll_ready(cx)

295 }

296

297 fn call(&mut self, request: Request<B>) -> Self::Future {

298 let span = http_span(&request);

299 self.inner.call(request).instrument(span)

300 }

301}

302

303/// Client span for an outbound HTTP call; the trace context of this span is

304/// what gets injected into the request, so the remote hop parents to it.

305pub fn http_client_span(method: &str, url: &str) -> Span {

306 let name = format!("{method} {url}");

307 tracing::info_span!(

308 "http.client",

309 "otel.name" = %name,

310 "otel.kind" = "client",

311 "http.request.method" = %method,

312 "url.full" = %url,

313 "http.response.status_code" = tracing::field::Empty,

314 )

315}

316

317/// Sends a request inside `span` with `traceparent` injected into its headers.

318pub async fn send(

319 span: Span,

320 request: reqwest::RequestBuilder,

321) -> Result<reqwest::Response, reqwest::Error> {

322 let mut headers = HeaderMap::new();

323 global::get_text_map_propagator(|propagator| {

324 propagator.inject_context(&span.context(), &mut HeaderInjector(&mut headers));

325 });

326 if let Some(request_id) = crate::middleware::request_id()

327 && let Ok(value) = request_id.parse()

328 {

329 headers.insert(http::header::HeaderName::from_static("x-request-id"), value);

330 }

331

332 let result = request

333 .headers(headers)

334 .send()

335 .instrument(span.clone())

336 .await;

337 match &result {

338 Ok(response) => {

339 span.record(

340 "http.response.status_code",

341 i64::from(response.status().as_u16()),

342 );

343 }

344 Err(_) => {

345 span.record("otel.status_code", "error");

346 }

347 }

348 result

349}

350

351pub async fn shutdown_signal() {

352 let ctrl_c = async {

353 if let Err(err) = tokio::signal::ctrl_c().await {

354 warn!(error = %err, "failed to install ctrl_c handler");

355 }

356 };

357

358 #[cfg(unix)]

359 let terminate = async {

360 use tokio::signal::unix::{SignalKind, signal};

361 match signal(SignalKind::terminate()) {

362 Ok(mut s) => {

363 s.recv().await;

364 }

365 Err(err) => {

366 warn!(error = %err, "failed to install SIGTERM handler");

367 }

368 }

369 };

370

371 #[cfg(not(unix))]

372 let terminate = std::future::pending::<()>();

373

374 tokio::select! {

375 _ = ctrl_c => info!("received SIGINT"),

376 _ = terminate => info!("received SIGTERM"),

377 }

378

379 shutdown_telemetry();

380}

381

382/// Drains spans queued in the batch processor and metrics queued in the

383/// periodic reader. Without this the tail of every deployment's telemetry is

384/// lost on rollout, which is exactly when you look at it.

385pub fn shutdown_telemetry() {

386 flush_traces();

387 if let Some(provider) = METER_PROVIDER.get()

388 && let Err(err) = provider.shutdown()

389 {

390 warn!(error = %err, "failed to shut down meter provider");

391 }

392}

393

394pub fn flush_traces() {

395 if let Some(provider) = TRACER_PROVIDER.get()

396 && let Err(err) = provider.shutdown()

397 {

398 warn!(error = %err, "failed to flush tracer provider");

399 }

400}

Modified lib/export/Cargo.toml +1 −9 View lib/export/Cargo.toml

@@ -20,19 +20,11 @@

2020imaging = { workspace = true }

2121db = { workspace = true }

2222actix-web = { workspace = true }

23http = "1.5.0"

2423serde = { workspace = true, features = ["derive"] }

2524sha2 = { workspace = true }

2625tracing = { workspace = true }

27tracing-subscriber = { workspace = true }

28opentelemetry = { workspace = true }

29opentelemetry_sdk = { workspace = true }

30opentelemetry-http = { workspace = true }

31opentelemetry-otlp = { workspace = true }

32reqwest = { workspace = true }

33tracing-opentelemetry = { workspace = true }

26oltp = { workspace = true }

3427utoipa = { workspace = true }

35tower = { workspace = true }

3628serde_json = { workspace = true }

3729uuid = { workspace = true }

3830bytes = { workspace = true }

Added lib/oltp/src/client.rs +52 −0 View lib/oltp/src/client.rs

@@ -0,0 +1,52 @@

1use opentelemetry::global;

2use opentelemetry_http::HeaderInjector;

3use tracing::{Instrument as _, Span};

4use tracing_opentelemetry::OpenTelemetrySpanExt;

5

6/// Client span for an outbound HTTP call; the trace context of this span is

7/// what gets injected into the request, so the remote hop parents to it.

8pub fn http_client_span(method: &str, url: &str) -> Span {

9 let name = format!("{method} {url}");

10 tracing::info_span!(

11 "http.client",

12 "otel.name" = %name,

13 "otel.kind" = "client",

14 "http.request.method" = %method,

15 "url.full" = %url,

16 "http.response.status_code" = tracing::field::Empty,

17 )

18}

19

20/// Sends a request inside `span` with `traceparent` injected into its headers.

21pub async fn send(

22 span: Span,

23 request: reqwest::RequestBuilder,

24) -> Result<reqwest::Response, reqwest::Error> {

25 let mut headers = http::HeaderMap::new();

26 global::get_text_map_propagator(|propagator| {

27 propagator.inject_context(&span.context(), &mut HeaderInjector(&mut headers));

28 });

29 if let Some(request_id) = envelope::context::request_id()

30 && let Ok(value) = request_id.parse()

31 {

32 headers.insert(http::header::HeaderName::from_static("x-request-id"), value);

33 }

34

35 let result = request

36 .headers(headers)

37 .send()

38 .instrument(span.clone())

39 .await;

40 match &result {

41 Ok(response) => {

42 span.record(

43 "http.response.status_code",

44 i64::from(response.status().as_u16()),

45 );

46 }

47 Err(_) => {

48 span.record("otel.status_code", "error");

49 }

50 }

51 result

52}

Modified lib/oltp/src/lib.rs +1 −0 View lib/oltp/src/lib.rs

@@ -1,4 +1,5 @@

11pub mod actix;

2pub mod client;

23pub mod config;

34pub mod error;

45pub mod grpc;

Modified lib/oltp/src/metrics.rs +99 −2 View lib/oltp/src/metrics.rs

@@ -1,7 +1,7 @@

1use std::sync::OnceLock;

1use std::sync::{LazyLock, OnceLock};

22use std::time::Instant;

33

4use opentelemetry::metrics::{Counter, Histogram, UpDownCounter};

4use opentelemetry::metrics::{Counter, Gauge, Histogram, Meter, UpDownCounter};

55use opentelemetry::{KeyValue, global};

66use opentelemetry_otlp::{MetricExporter, WithExportConfig, WithHttpConfig};

77use opentelemetry_sdk::Resource;

@@ -145,3 +145,100 @@

145145 }

146146 builder.build()

147147}

148

149fn app_meter() -> Meter {

150 global::meter("gitdata")

151}

152

153static AUTH_LOGINS: LazyLock<Counter<u64>> = LazyLock::new(|| {

154 app_meter()

155 .u64_counter("auth.logins")

156 .with_description("Login attempts by outcome")

157 .build()

158});

159

160static GIT_OPS: LazyLock<Counter<u64>> = LazyLock::new(|| {

161 app_meter()

162 .u64_counter("git.ops")

163 .with_description("Git protocol operations by operation and outcome")

164 .build()

165});

166

167static PUSH_DELIVERIES: LazyLock<Counter<u64>> = LazyLock::new(|| {

168 app_meter()

169 .u64_counter("push.deliveries")

170 .with_description("Web push deliveries by outcome")

171 .build()

172});

173

174static IM_CONNECTIONS: LazyLock<UpDownCounter<i64>> = LazyLock::new(|| {

175 app_meter()

176 .i64_up_down_counter("im.connections")

177 .with_description("Open chat WebSocket connections")

178 .build()

179});

180

181static CHAT_WEBHOOK_DELIVERIES: LazyLock<Counter<u64>> = LazyLock::new(|| {

182 app_meter()

183 .u64_counter("chat.webhook.deliveries")

184 .with_description("Outbound chat webhook deliveries by outcome")

185 .build()

186});

187

188static QUEUE_DEPTH: LazyLock<Gauge<i64>> = LazyLock::new(|| {

189 app_meter()

190 .i64_gauge("queue.depth")

191 .with_description("Pending queue entries by topic")

192 .build()

193});

194

195pub fn auth_login(outcome: &'static str) {

196 AUTH_LOGINS.add(1, &[KeyValue::new("outcome", outcome)]);

197}

198

199pub fn git_op(op: &'static str, outcome: &'static str) {

200 GIT_OPS.add(

201 1,

202 &[KeyValue::new("op", op), KeyValue::new("outcome", outcome)],

203 );

204}

205

206pub fn push_delivery(outcome: &'static str) {

207 PUSH_DELIVERIES.add(1, &[KeyValue::new("outcome", outcome)]);

208}

209

210pub fn im_connection(delta: i64) {

211 IM_CONNECTIONS.add(delta, &[]);

212}

213

214/// `Drop`-based guard so every socket exit path (incl. early returns) balances

215/// the connection gauge.

216pub struct ImConnectionGuard;

217

218impl ImConnectionGuard {

219 #[must_use]

220 pub fn new() -> Self {

221 im_connection(1);

222 Self

223 }

224}

225

226impl Default for ImConnectionGuard {

227 fn default() -> Self {

228 Self::new()

229 }

230}

231

232impl Drop for ImConnectionGuard {

233 fn drop(&mut self) {

234 im_connection(-1);

235 }

236}

237

238pub fn chat_webhook_delivery(outcome: &'static str) {

239 CHAT_WEBHOOK_DELIVERIES.add(1, &[KeyValue::new("outcome", outcome)]);

240}

241

242pub fn queue_depth(topic: &'static str, depth: i64) {

243 QUEUE_DEPTH.record(depth, &[KeyValue::new("topic", topic)]);

244}

Modified lib/oltp/Cargo.toml +4 −0 View lib/oltp/Cargo.toml

@@ -12,8 +12,12 @@

1212http-body = "1.1.0"

1313opentelemetry = { version = "0.32.0" }

1414opentelemetry-appender-tracing = { version = "0.32.0", features = ["experimental_span_attributes"] }

15# reqwest-rustls（见下）：oltp 的 OTLP exporter 走 reqwest，默认特性不带 TLS 后端，

16# https 的 OTEL_EXPORTER_OTLP_ENDPOINT 会每次导出都失败，靠依赖里的 reqwest(rustls) 统一特性补齐。

1517opentelemetry-otlp = { version = "0.32.0", default-features = false, features = ["http-proto", "reqwest-blocking-client", "internal-logs", "logs", "metrics", "trace"] }

1618opentelemetry_sdk = { version = "0.32.1" }

19opentelemetry-http = { workspace = true }

20reqwest = { workspace = true }

1721thiserror = { workspace = true }

1822tokio = { workspace = true, features = ["time"] }

1923tonic = { workspace = true }

Modified lib/push/src/client.rs +2 −2 View lib/push/src/client.rs

@@ -140,8 +140,8 @@

140140 for (name, value) in request.headers() {

141141 outbound = outbound.header(name.as_str(), value.as_bytes());

142142 }

143 let span = export::telemetry::http_client_span("POST", &request.uri().to_string());

144 let response = export::telemetry::send(span, outbound.body(request.into_body()))

143 let span = oltp::client::http_client_span("POST", &request.uri().to_string());

144 let response = oltp::client::send(span, outbound.body(request.into_body()))

145145 .await

146146 .map_err(|error| PushError::Transport(error.to_string()))?;

147147

Modified lib/push/src/worker.rs +7 −7 View lib/push/src/worker.rs

@@ -90,7 +90,7 @@

9090 if let Some(request_id) = job.request_id.as_deref() {

9191 span.record("request_id", request_id);

9292 }

93 export::telemetry::set_parent_from_traceparent(&span, job.trace.as_deref());

93 oltp::propagation::set_parent_from_traceparent(&span, job.trace.as_deref());

9494 if let Err(error) = export::middleware::with_request_id(

9595 job.request_id.as_deref(),

9696 self.handle(&job).instrument(span),

@@ -141,11 +141,11 @@

141141 };

142142 match deliver(&self.client, &self.keys, &keys, &payload, ttl_secs).await {

143143 Ok(DeliveryOutcome::Delivered) => {

144 export::metrics::push_delivery("delivered");

144 oltp::metrics::push_delivery("delivered");

145145 tracing::debug!(user = %job.user_id, "push delivered");

146146 }

147147 Ok(DeliveryOutcome::Gone) => {

148 export::metrics::push_delivery("gone");

148 oltp::metrics::push_delivery("gone");

149149 // 用户清了浏览器数据或卸载了站点：一次性清理，不是错误。

150150 tracing::debug!(user = %job.user_id, "push subscription expired; removing");

151151 if let Err(error) =

@@ -157,7 +157,7 @@

157157 }

158158 }

159159 Ok(DeliveryOutcome::Rejected) => {

160 export::metrics::push_delivery("rejected");

160 oltp::metrics::push_delivery("rejected");

161161 tracing::warn!(

162162 user = %job.user_id,

163163 endpoint = %subscription.endpoint,

@@ -168,11 +168,11 @@

168168 return Ok(());

169169 }

170170 Ok(DeliveryOutcome::Retry) => {

171 export::metrics::push_delivery("retry");

171 oltp::metrics::push_delivery("retry");

172172 retry = true;

173173 }

174174 Err(PushError::Subscription(message)) => {

175 export::metrics::push_delivery("subscription_error");

175 oltp::metrics::push_delivery("subscription_error");

176176 // 订阅本身不可修（endpoint 或密钥解析失败），重投没有意义。

177177 tracing::warn!(user = %job.user_id, %message, "dropping broken subscription");

178178 if let Err(error) =

@@ -184,7 +184,7 @@

184184 }

185185 }

186186 Err(error) => {

187 export::metrics::push_delivery("error");

187 oltp::metrics::push_delivery("error");

188188 tracing::warn!(%error, user = %job.user_id, "push delivery error");

189189 retry = true;

190190 }

Modified lib/push/Cargo.toml +1 −0 View lib/push/Cargo.toml

@@ -7,6 +7,7 @@

77config = { workspace = true }

88export = { workspace = true }

99models = { workspace = true }

10oltp = { workspace = true }

1011base64 = { workspace = true }

1112serde = { workspace = true, features = ["derive"] }

1213serde_json = { workspace = true }

Modified Cargo.lock +6 −9 View Cargo.lock

@@ -1271,6 +1271,7 @@

12711271 "livekit-api",

12721272 "livekit-protocol",

12731273 "models",

1274 "oltp",

12741275 "queue",

12751276 "rand 0.8.8",

12761277 "redis",

@@ -2339,18 +2340,13 @@

23392340 "email",

23402341 "envelope",

23412342 "events",

2342 "http 1.5.0",

23432343 "imaging",

23442344 "models",

2345 "opentelemetry",

2346 "opentelemetry-http",

2347 "opentelemetry-otlp",

2348 "opentelemetry_sdk",

2345 "oltp",

23492346 "prost-types 0.14.4",

23502347 "proto",

23512348 "queue",

23522349 "redis",

2353 "reqwest 0.12.28",

23542350 "sea-orm",

23552351 "serde",

23562352 "serde_json",

@@ -2361,10 +2357,7 @@

23612357 "time 0.3.55",

23622358 "tokio",

23632359 "tonic",

2364 "tower",

23652360 "tracing",

2366 "tracing-opentelemetry",

2367 "tracing-subscriber",

23682361 "utoipa 5.5.0",

23692362 "uuid",

23702363]

@@ -3813,6 +3806,7 @@

38133806 "events",

38143807 "export",

38153808 "models",

3809 "oltp",

38163810 "push",

38173811 "queue",

38183812 "sea-orm",

@@ -4756,8 +4750,10 @@

47564750 "http-body 1.1.0",

47574751 "opentelemetry",

47584752 "opentelemetry-appender-tracing",

4753 "opentelemetry-http",

47594754 "opentelemetry-otlp",

47604755 "opentelemetry_sdk",

4756 "reqwest 0.12.28",

47614757 "session",

47624758 "thiserror 2.0.21",

47634759 "tokio",

@@ -5709,6 +5705,7 @@

57095705 "export",

57105706 "http 1.5.0",

57115707 "models",

5708 "oltp",

57125709 "reqwest 0.12.28",

57135710 "sea-orm",

57145711 "serde",

Modified Cargo.toml +0 −5 View Cargo.toml

@@ -51,12 +51,7 @@

5151actix-cors = "0.7"

5252tower = "0.5.3"

5353opentelemetry = "0.32"

54opentelemetry_sdk = { version = "0.32.1", features = ["rt-tokio", "trace"] }

5554opentelemetry-http = "0.32"

56# reqwest-rustls: the default client feature set compiles reqwest with no TLS backend,

57# so an https OTEL_EXPORTER_OTLP_ENDPOINT would fail every export at request time.

58opentelemetry-otlp = { version = "0.32", features = ["reqwest-rustls"] }

59tracing-opentelemetry = "0.33"

6055image = { version = "0.25", default-features = false, features = ["jpeg", "png", "webp", "gif"] }

6156tracing = "0.1.44"

6257figment = { version = "0.10", features = ["toml", "env"] }