feat(metrics): export HTTP and business metrics from every host

Modified app/aihost/src/main.rs +15 −0 View app/aihost/src/main.rs

@@ -28,6 +28,16 @@

2828 let state = export::AppState::with_config(cfg.clone().into_app_config())

2929 .await

3030 .context("building AppState")?;

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

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

33 }

34 export::metrics::spawn_queue_depth_reporter(

35 state.redis.clone(),

36 &[

37 ("ai.jobs", cfg.ai.queue_topic.clone()),

38 ("ai.jobs.dlq", cfg.ai.dlq_topic.clone()),

39 ],

40 );

3141 let service = AgentService::new(state).context("building ai service")?;

3242 info!(model = service.client().model(), "dependencies ready");

3343

@@ -108,14 +118,18 @@

108118 let job_id = task.job_id();

109119 let span = tracing::info_span!("ai.task", job_id = %job_id, kind = task.kind_str());

110120 export::telemetry::set_parent_from_traceparent(&span, task.trace());

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

111122 match service.run_task(&task).instrument(span).await {

112123 Ok(()) => {

113124 attempts.remove(&job_id);

125 export::metrics::ai_job(task.kind_str(), "ok");

126 export::metrics::ai_job_duration(task.kind_str(), started.elapsed());

114127 info!(%job_id, kind = task.kind_str(), "ai task completed");

115128 }

116129 Err(error) => {

117130 if !error.is_retryable() {

118131 attempts.remove(&job_id);

132 export::metrics::ai_job(task.kind_str(), "failed");

119133 warn!(%error, %job_id, "ai task is not retryable; sending to DLQ");

120134 service.mark_failed(job_id, &error.to_string()).await;

121135 let _ = queue.requeue(&dlq, &payload).await;

@@ -125,6 +139,7 @@

125139 *count += 1;

126140 if *count >= max_attempts {

127141 attempts.remove(&job_id);

142 export::metrics::ai_job(task.kind_str(), "failed");

128143 warn!(%error, %job_id, "ai task failed permanently; sending to DLQ");

129144 service.mark_failed(job_id, &error.to_string()).await;

130145 let _ = queue.requeue(&dlq, &payload).await;

Modified app/apihost/src/main.rs +7 −0 View app/apihost/src/main.rs

@@ -28,6 +28,13 @@

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

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

3030 }

31 export::metrics::spawn_queue_depth_reporter(

32 state.redis.clone(),

33 &[

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

35 ("push.jobs.dlq", export::push::DLQ_TOPIC.to_string()),

36 ],

37 );

3138

3239 tokio::spawn(chat::worker::run_background_tasks(state.clone()));

3340 tokio::spawn(auth::worker::run_deletion_executor(state.clone()));

Modified app/githost/src/main.rs +4 −0 View app/githost/src/main.rs

@@ -22,6 +22,10 @@

2222 .context("building AppState")?;

2323 info!("dependencies ready");

2424

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

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

27 }

28

2529 let git_svc = git::service::GitService::new(state.clone());

2630 let sync_queue = Arc::new(git::sync::queue::SyncQueue::new(state.redis.clone()));

2731 let smart_svc = git::smart::service::SmartService::new(git_svc)

Modified app/im/src/main.rs +4 −0 View app/im/src/main.rs

@@ -29,6 +29,10 @@

2929 .context("building AppState")?;

3030 info!("dependencies ready");

3131

32 if let Err(error) = export::telemetry::init_meter_provider() {

33 tracing::warn!(%error, "failed to init meter provider; metrics disabled");

34 }

35

3236 let chat = ChatService::new(state);

3337

3438 let bind_addr = cfg.im.bind_addr().to_string();

Modified app/im/src/socket.rs +1 −0 View app/im/src/socket.rs

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

5252}

5353

5454pub async fn handle_socket(socket: ServerWs, service: ChatService, uid: Uuid, session_id: Uuid) {

55 let _connection = export::metrics::ImConnectionGuard::new();

5556 let (mut sink, mut stream) = socket.split();

5657

5758 let (inbox_tx, mut inbox_rx) = mpsc::channel::<Inbox>(1024);

Modified app/mails/src/main.rs +4 −0 View app/mails/src/main.rs

@@ -31,6 +31,10 @@

3131 "starting mails worker"

3232 );

3333

34 if let Err(error) = export::telemetry::init_meter_provider() {

35 tracing::warn!(%error, "failed to init meter provider; metrics disabled");

36 }

37

3438 let opts = WorkerOptions {

3539 template_dir: None,

3640 shutdown_timeout: Some(Duration::from_secs(5)),

Modified app/mcp/src/main.rs +6 −1 View app/mcp/src/main.rs

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

77use tokio::net::TcpListener;

88use tower_http::limit::RequestBodyLimitLayer;

99use tower_http::trace::TraceLayer;

10use tracing::info;

10use tracing::{info, warn};

1111

1212const MAX_REQUEST_BODY_BYTES: usize = 2 * 1024 * 1024;

1313

@@ -27,6 +27,10 @@

2727 .context("building AppState")?;

2828 info!("dependencies ready");

2929

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

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

32 }

33

3034 let health = export::health::router_with(Some(state.database.clone()), state.redis.clone());

3135 let mcp = mcp_api::router(state).await;

3236

@@ -40,6 +44,7 @@

4044 .make_span_with(export::telemetry::HttpSpan)

4145 .on_response(export::telemetry::HttpOnResponse),

4246 )

47 .layer(export::telemetry::metrics_layer())

4348 .layer(export::middleware::AppTraceLayer::new());

4449

4550 let bind_addr = cfg.mcp.bind_addr().to_string();

Modified app/synchost/src/main.rs +11 −0 View app/synchost/src/main.rs

@@ -31,6 +31,17 @@

3131 let repo_root = PathBuf::from(cfg.git.repo_root.clone());

3232 info!("dependencies ready");

3333

34 if let Err(error) = export::telemetry::init_meter_provider() {

35 tracing::warn!(%error, "failed to init meter provider; metrics disabled");

36 }

37 export::metrics::spawn_queue_depth_reporter(

38 redis.clone(),

39 &[

40 ("git.sync", cfg.git.sync.queue_topic.clone()),

41 ("git.sync.dlq", cfg.git.sync.dlq_topic.clone()),

42 ],

43 );

44

3445 let cancel = CancellationToken::new();

3546

3647 let search = SearchService::from_parts(db.clone(), cfg.search.clone())

Modified app/webhook/src/main.rs +20 −1 View app/webhook/src/main.rs

@@ -26,6 +26,17 @@

2626 let service = CiService::new(state);

2727 info!("dependencies ready");

2828

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

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

31 }

32 export::metrics::spawn_queue_depth_reporter(

33 service.state.redis.clone(),

34 &[

35 ("ci.events", cfg.ci.event_queue_topic.clone()),

36 ("ci.events.dlq", cfg.ci.event_dlq_topic.clone()),

37 ],

38 );

39

2940 let cancel = CancellationToken::new();

3041

3142 let consumer = if cfg.ci.enabled {

@@ -70,7 +81,8 @@

7081 Some(service.state.database.clone()),

7182 service.state.redis.clone(),

7283 ))

73 .layer(export::telemetry::http_trace_layer());

84 .layer(export::telemetry::http_trace_layer())

85 .layer(export::telemetry::metrics_layer());

7486 info!(%addr, "webhook scheduler listening");

7587

7688 let server_cancel = cancel.clone();

@@ -128,6 +140,12 @@

128140 continue;

129141 }

130142 };

143 let event_kind = match &event {

144 ci_protocol::events::CiEvent::Push(_) => "push",

145 ci_protocol::events::CiEvent::PullRequest(_) => "pull_request",

146 ci_protocol::events::CiEvent::Trigger(_) => "trigger",

147 };

148 export::metrics::ci_event(event_kind);

131149 let event_id = match &event {

132150 ci_protocol::events::CiEvent::Push(push) => push.id,

133151 ci_protocol::events::CiEvent::PullRequest(pr) => pr.id,

@@ -138,6 +156,7 @@

138156 match service.handle_event(&event).instrument(span).await {

139157 Ok(created) => {

140158 attempts.remove(&event_id);

159 export::metrics::ci_runs_created(event_kind, created.len());

141160 if !created.is_empty() {

142161 info!(event = %event_id, runs = created.len(), "ci event produced runs");

143162 }

Modified lib/agent/src/queue.rs +1 −0 View lib/agent/src/queue.rs

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

5454 prompt_tokens: Option<i32>,

5555 completion_tokens: Option<i32>,

5656 ) {

57 export::metrics::ai_tokens(prompt_tokens, completion_tokens);

5758 let db = self.db_write();

5859 let now = OffsetDateTime::now_utc();

5960 if let Ok(Some(row)) = ai_job::Entity::find_by_id(job_id).one(db).await {

Modified lib/auth/src/services/login.rs +29 −0 View lib/auth/src/services/login.rs

@@ -63,6 +63,22 @@

6363 user_agent: Option<&str>,

6464 client_ip: Option<&str>,

6565 ) -> Result<LoginResp, AuthError> {

66 let result = self.login_inner(session, cmd, user_agent, client_ip).await;

67 export::metrics::auth_login(match &result {

68 Ok(resp) if resp.needs_2fa => "challenge",

69 Ok(_) => "ok",

70 Err(_) => "rejected",

71 });

72 result

73 }

74

75 async fn login_inner(

76 &self,

77 session: &Session,

78 cmd: LoginCmd,

79 user_agent: Option<&str>,

80 client_ip: Option<&str>,

81 ) -> Result<LoginResp, AuthError> {

6682 self.verify_pow(session, &cmd.pow_answer)?;

6783

6884 let ua_hash = user_agent.map(|ua| sha256_b64url(ua.as_bytes()));

@@ -132,6 +148,19 @@

132148 session: &Session,

133149 cmd: TwoFactorLoginCmd,

134150 ) -> Result<LoginResp, AuthError> {

151 let result = self.login_2fa_inner(session, cmd).await;

152 export::metrics::auth_login(match &result {

153 Ok(_) => "ok",

154 Err(_) => "rejected",

155 });

156 result

157 }

158

159 async fn login_2fa_inner(

160 &self,

161 session: &Session,

162 cmd: TwoFactorLoginCmd,

163 ) -> Result<LoginResp, AuthError> {

135164 let Some(challenge) = session.get::<TwoFactorChallenge>(TWO_FA_KEY) else {

136165 return Err(AuthError::BadRequest(

137166 "no pending two-factor challenge".to_string(),

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

@@ -552,11 +552,16 @@

552552 .body(body),

553553 )

554554 .await

555 .map_err(|e| format!("request failed: {e}"))?;

555 .map_err(|e| {

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

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

558 })?;

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

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

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

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

559563 } else {

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

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

561566 }

562567 }

Modified lib/ci/src/outbound.rs +6 −1 View lib/ci/src/outbound.rs

@@ -239,7 +239,10 @@

239239 .body(body),

240240 )

241241 .await

242 .map_err(|e| format!("request failed: {e}"))?;

242 .map_err(|e| {

243 export::metrics::ci_webhook_delivery("error");

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

245 })?;

243246 let status = response.status().as_u16() as i32;

244247 let text = response

245248 .text()

@@ -249,8 +252,10 @@

249252 .take(MAX_RESPONSE_BODY)

250253 .collect::<String>();

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

255 export::metrics::ci_webhook_delivery("ok");

252256 Ok((status, text))

253257 } else {

258 export::metrics::ci_webhook_delivery("error");

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

255260 }

256261 }

Modified lib/ci/src/scheduler.rs +4 −0 View lib/ci/src/scheduler.rs

@@ -963,6 +963,10 @@

963963 if updated.rows_affected == 0 {

964964 return Ok(job.run_id);

965965 }

966 let duration = job

967 .started_at

968 .and_then(|started| (now - started).try_into().ok());

969 export::metrics::ci_job_finished(state.as_str(), duration);

966970 if state != WorkflowJobState::Success {

967971 workflow_step::Entity::update_many()

968972 .col_expr(

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

@@ -1,5 +1,6 @@

11pub mod error;

22pub mod events;

3pub mod metrics;

34pub mod queue;

45pub mod service;

56pub mod subscriber;

Added lib/email/src/metrics.rs +16 −0 View lib/email/src/metrics.rs

@@ -0,0 +1,16 @@

1use std::sync::LazyLock;

2

3use opentelemetry::KeyValue;

4use opentelemetry::global;

5use opentelemetry::metrics::Counter;

6

7static EMAIL_SENT: LazyLock<Counter<u64>> = LazyLock::new(|| {

8 global::meter("gitdata")

9 .u64_counter("email.sent")

10 .with_description("Email sends by outcome")

11 .build()

12});

13

14pub fn email_sent(outcome: &'static str) {

15 EMAIL_SENT.add(1, &[KeyValue::new("outcome", outcome)]);

16}

Modified lib/email/src/service.rs +10 −1 View lib/email/src/service.rs

@@ -76,7 +76,16 @@

7676 );

7777

7878 let message = build_message(&req, from, self.inner.max_attachment_bytes)?;

79 let smtp_message_id = transport.send(message).await?;

79 let smtp_message_id = match transport.send(message).await {

80 Ok(message_id) => {

81 crate::metrics::email_sent("ok");

82 message_id

83 }

84 Err(error) => {

85 crate::metrics::email_sent("error");

86 return Err(error);

87 }

88 };

8089 let sent = EmailSent {

8190 request_id: req.request_id,

8291 recipient_count,

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

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

55pub mod health;

66pub mod imaging;

77pub mod jwt;

8pub mod metrics;

89pub mod middleware;

910pub mod notify;

1011pub mod pagination;

Added lib/export/src/metrics.rs +284 −0 View lib/export/src/metrics.rs

@@ -0,0 +1,284 @@

1use std::sync::LazyLock;

2use std::time::Duration;

3

4use opentelemetry::global;

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

6use opentelemetry::{KeyValue, metrics::Meter};

7

8use crate::redis::AppRedis;

9

10fn meter() -> Meter {

11 global::meter("gitdata")

12}

13

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

15 meter()

16 .u64_counter("auth.logins")

17 .with_description("Login attempts by outcome")

18 .build()

19});

20

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

22 meter()

23 .u64_counter("git.ops")

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

25 .build()

26});

27

28static CI_EVENTS: LazyLock<Counter<u64>> = LazyLock::new(|| {

29 meter()

30 .u64_counter("ci.events")

31 .with_description("CI events consumed by kind")

32 .build()

33});

34

35static CI_RUNS_CREATED: LazyLock<Counter<u64>> = LazyLock::new(|| {

36 meter()

37 .u64_counter("ci.runs.created")

38 .with_description("Workflow runs created from events")

39 .build()

40});

41

42static CI_JOBS_FINISHED: LazyLock<Counter<u64>> = LazyLock::new(|| {

43 meter()

44 .u64_counter("ci.jobs.finished")

45 .with_description("Runner jobs finished by conclusion")

46 .build()

47});

48

49static CI_JOB_DURATION: LazyLock<Histogram<f64>> = LazyLock::new(|| {

50 meter()

51 .f64_histogram("ci.job.duration")

52 .with_unit("s")

53 .with_description("Runner job wall-clock duration")

54 .build()

55});

56

57static CI_WEBHOOK_DELIVERIES: LazyLock<Counter<u64>> = LazyLock::new(|| {

58 meter()

59 .u64_counter("ci.webhook.deliveries")

60 .with_description("Outbound CI webhook deliveries by outcome")

61 .build()

62});

63

64static AI_JOBS: LazyLock<Counter<u64>> = LazyLock::new(|| {

65 meter()

66 .u64_counter("ai.jobs")

67 .with_description("AI jobs processed by kind and outcome")

68 .build()

69});

70

71static AI_JOB_DURATION: LazyLock<Histogram<f64>> = LazyLock::new(|| {

72 meter()

73 .f64_histogram("ai.job.duration")

74 .with_unit("s")

75 .with_description("AI job wall-clock duration")

76 .build()

77});

78

79static AI_TOKENS: LazyLock<Counter<u64>> = LazyLock::new(|| {

80 meter()

81 .u64_counter("ai.tokens")

82 .with_description("Model tokens consumed by type")

83 .build()

84});

85

86static EMAIL_SENT: LazyLock<Counter<u64>> = LazyLock::new(|| {

87 meter()

88 .u64_counter("email.sent")

89 .with_description("Email sends by outcome")

90 .build()

91});

92

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

94 meter()

95 .u64_counter("push.deliveries")

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

97 .build()

98});

99

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

101 meter()

102 .i64_up_down_counter("im.connections")

103 .with_description("Open chat WebSocket connections")

104 .build()

105});

106

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

108 meter()

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

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

111 .build()

112});

113

114static SYNC_JOBS: LazyLock<Counter<u64>> = LazyLock::new(|| {

115 meter()

116 .u64_counter("git.sync.jobs")

117 .with_description("Repository sync jobs by outcome")

118 .build()

119});

120

121static IMPORT_JOBS: LazyLock<Counter<u64>> = LazyLock::new(|| {

122 meter()

123 .u64_counter("import.jobs")

124 .with_description("Repository import jobs by outcome")

125 .build()

126});

127

128static SEARCH_JOBS: LazyLock<Counter<u64>> = LazyLock::new(|| {

129 meter()

130 .u64_counter("search.index.jobs")

131 .with_description("Search index jobs by outcome")

132 .build()

133});

134

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

136 meter()

137 .i64_gauge("queue.depth")

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

139 .build()

140});

141

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

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

144}

145

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

147 GIT_OPS.add(

148 1,

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

150 );

151}

152

153pub fn ci_event(kind: &'static str) {

154 CI_EVENTS.add(1, &[KeyValue::new("kind", kind)]);

155}

156

157pub fn ci_runs_created(event: &'static str, count: usize) {

158 if count > 0 {

159 CI_RUNS_CREATED.add(count as u64, &[KeyValue::new("event", event)]);

160 }

161}

162

163pub fn ci_job_finished(conclusion: &'static str, duration: Option<Duration>) {

164 CI_JOBS_FINISHED.add(1, &[KeyValue::new("conclusion", conclusion)]);

165 if let Some(duration) = duration {

166 CI_JOB_DURATION.record(

167 duration.as_secs_f64(),

168 &[KeyValue::new("conclusion", conclusion)],

169 );

170 }

171}

172

173pub fn ci_webhook_delivery(outcome: &'static str) {

174 CI_WEBHOOK_DELIVERIES.add(1, &[KeyValue::new("outcome", outcome)]);

175}

176

177pub fn ai_job(kind: &str, outcome: &'static str) {

178 AI_JOBS.add(

179 1,

180 &[

181 KeyValue::new("kind", kind.to_string()),

182 KeyValue::new("outcome", outcome),

183 ],

184 );

185}

186

187pub fn ai_job_duration(kind: &str, duration: Duration) {

188 AI_JOB_DURATION.record(

189 duration.as_secs_f64(),

190 &[KeyValue::new("kind", kind.to_string())],

191 );

192}

193

194pub fn ai_tokens(prompt: Option<i32>, completion: Option<i32>) {

195 if let Some(prompt) = prompt {

196 AI_TOKENS.add(

197 u64::try_from(prompt).unwrap_or(0),

198 &[KeyValue::new("type", "prompt")],

199 );

200 }

201 if let Some(completion) = completion {

202 AI_TOKENS.add(

203 u64::try_from(completion).unwrap_or(0),

204 &[KeyValue::new("type", "completion")],

205 );

206 }

207}

208

209pub fn email_sent(outcome: &'static str) {

210 EMAIL_SENT.add(1, &[KeyValue::new("outcome", outcome)]);

211}

212

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

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

215}

216

217pub fn im_connection(delta: i64) {

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

219}

220

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

222/// the connection gauge.

223pub struct ImConnectionGuard;

224

225impl ImConnectionGuard {

226 #[must_use]

227 pub fn new() -> Self {

228 im_connection(1);

229 Self

230 }

231}

232

233impl Default for ImConnectionGuard {

234 fn default() -> Self {

235 Self::new()

236 }

237}

238

239impl Drop for ImConnectionGuard {

240 fn drop(&mut self) {

241 im_connection(-1);

242 }

243}

244

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

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

247}

248

249pub fn sync_job(outcome: &'static str) {

250 SYNC_JOBS.add(1, &[KeyValue::new("outcome", outcome)]);

251}

252

253pub fn import_job(outcome: &'static str) {

254 IMPORT_JOBS.add(1, &[KeyValue::new("outcome", outcome)]);

255}

256

257pub fn search_index_job(outcome: &'static str) {

258 SEARCH_JOBS.add(1, &[KeyValue::new("outcome", outcome)]);

259}

260

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

262

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

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

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

266 if queues.is_empty() {

267 return;

268 }

269 tokio::spawn(async move {

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

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

272 loop {

273 ticker.tick().await;

274 for (topic, key) in &queues {

275 match redis.llen(key).await {

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

277 Err(error) => {

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

279 }

280 }

281 }

282 }

283 });

284}

Modified lib/export/src/redis.rs +11 −0 View lib/export/src/redis.rs

@@ -161,6 +161,17 @@

161161 };

162162 result.map_err(ExportError::from)

163163 }

164

165 /// 队列积压深度（`LLEN`）；供指标采样使用。

166 pub async fn llen(&self, key: &str) -> Result<i64, ExportError> {

167 let mut conn = self.connection_clone();

168 let result: Result<i64, redis::RedisError> = match &mut conn {

169 RedisConnection::Standalone(c) => redis::cmd("LLEN").arg(key).query_async(c).await,

170 RedisConnection::Cluster(c) => redis::cmd("LLEN").arg(key).query_async(c).await,

171 };

172 result.map_err(ExportError::from)

173 }

174

164175 /// 推送队列是否有人消费；由 `AppState` 在构建时从 `[push] enabled` 写入。

165176 ///

166177 /// 放在 `AppRedis` 上是因为它是唯一"已经流到每个入队点"的句柄：否则要么

Modified lib/git/src/smart/handler.rs +13 −0 View lib/git/src/smart/handler.rs

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

2525 )

2626 .with_state(service)

2727 .layer(export::telemetry::http_trace_layer())

28 .layer(export::telemetry::metrics_layer())

2829}

2930

3031async fn dispatch(

@@ -135,6 +136,8 @@

135136 }

136137 _ => Ok(SmartError::BadRequest("method not allowed".into()).into_response()),

137138 };

139 let outcome = if result.is_ok() { "ok" } else { "error" };

140 export::metrics::git_op(op_of(&path_lower), outcome);

138141 result.unwrap_or_else(|e| {

139142 tracing::warn!(

140143 method = %method,

@@ -148,6 +151,16 @@

148151 })

149152}

150153

154fn op_of(path: &str) -> &'static str {

155 match path {

156 "info/refs" => "info_refs",

157 "git-upload-pack" => "upload_pack",

158 "git-receive-pack" => "receive_pack",

159 p if p.starts_with("info/lfs/") => "lfs",

160 _ => "objects",

161 }

162}

163

151164/// Whitelist of paths the GET fallback may serve. The dumb HTTP protocol

152165/// fetches `info/refs`, `HEAD`, `objects/info/packs`, `objects/pack/*`

153166/// and `objects/<xx>/<rest>`; everything else (notably `config`,

Modified lib/git/src/sync/worker.rs +3 −0 View lib/git/src/sync/worker.rs

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

209209

210210 match rsync.sync().instrument(span.clone()).await {

211211 Ok(()) => {

212 export::metrics::sync_job("ok");

212213 info!(

213214 consumer = id,

214215 task_id = %task.id,

@@ -239,6 +240,7 @@

239240 }

240241 }

241242 Err(e) if e.is_locked() => {

243 export::metrics::sync_job("locked");

242244 warn!(

243245 consumer = id,

244246 task_id = %task.id,

@@ -256,6 +258,7 @@

256258 }

257259 }

258260 Err(e) => {

261 export::metrics::sync_job("error");

259262 error!(

260263 consumer = id,

261264 task_id = %task.id,

Modified lib/import/src/worker.rs +5 −0 View lib/import/src/worker.rs

@@ -185,6 +185,11 @@

185185 let heartbeat = self.spawn_heartbeat(row.id);

186186 let outcome = self.run_job(&row).await;

187187 heartbeat.abort();

188 export::metrics::import_job(match &outcome {

189 Ok(_) => "ok",

190 Err(ImportError::Canceled) => "canceled",

191 Err(_) => "failed",

192 });

188193 let stage = ImportStage::parse(&row.stage);

189194 match outcome {

190195 Ok(stats) => {

Modified lib/push/src/worker.rs +9 −1 View lib/push/src/worker.rs

@@ -130,9 +130,11 @@

130130 };

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

132132 Ok(DeliveryOutcome::Delivered) => {

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

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

134135 }

135136 Ok(DeliveryOutcome::Gone) => {

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

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

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

138140 if let Err(error) =

@@ -144,6 +146,7 @@

144146 }

145147 }

146148 Ok(DeliveryOutcome::Rejected) => {

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

147150 tracing::warn!(

148151 user = %job.user_id,

149152 endpoint = %subscription.endpoint,

@@ -153,8 +156,12 @@

153156 self.forget(job);

154157 return Ok(());

155158 }

156 Ok(DeliveryOutcome::Retry) => retry = true,

159 Ok(DeliveryOutcome::Retry) => {

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

161 retry = true;

162 }

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

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

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

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

160167 if let Err(error) =

@@ -166,6 +173,7 @@

166173 }

167174 }

168175 Err(error) => {

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

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

170178 retry = true;

171179 }

Modified lib/search/src/worker.rs +1 −0 View lib/search/src/worker.rs

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

165165 }

166166 let started = tokio::time::Instant::now();

167167 let result = self.process(&job).await;

168 export::metrics::search_index_job(if result.is_ok() { "ok" } else { "error" });

168169 match result {

169170 Ok(()) => {

170171 index_job::Entity::update_many()